Alexa typing part 2 (#97911)

* Alexa typing part 2

* Update homeassistant/components/alexa/intent.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Missed type hints

* precision

* Follow up comment

* value

* revert abstract class changes

* raise NotImplementedError()

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Jan Bouwhuis 2023-08-07 23:26:44 +02:00 committed by GitHub
parent c8256d1d3d
commit 5657cfa052
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 95 additions and 69 deletions

View File

@ -3,7 +3,7 @@ from __future__ import annotations
from collections.abc import Generator, Iterable
import logging
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from homeassistant.components import (
alarm_control_panel,
@ -274,22 +274,22 @@ class AlexaEntity:
self.entity_conf = config.entity_config.get(entity.entity_id, {})
@property
def entity_id(self):
def entity_id(self) -> str:
"""Return the Entity ID."""
return self.entity.entity_id
def friendly_name(self):
def friendly_name(self) -> str:
"""Return the Alexa API friendly name."""
return self.entity_conf.get(CONF_NAME, self.entity.name).translate(
TRANSLATION_TABLE
)
def description(self):
def description(self) -> str:
"""Return the Alexa API description."""
description = self.entity_conf.get(CONF_DESCRIPTION) or self.entity_id
return f"{description} via Home Assistant".translate(TRANSLATION_TABLE)
def alexa_id(self):
def alexa_id(self) -> str:
"""Return the Alexa API entity id."""
return generate_alexa_id(self.entity.entity_id)
@ -317,7 +317,7 @@ class AlexaEntity:
"""
raise NotImplementedError
def serialize_properties(self):
def serialize_properties(self) -> Generator[dict[str, Any], None, None]:
"""Yield each supported property in API format."""
for interface in self.interfaces():
if not interface.properties_proactively_reported():
@ -325,9 +325,9 @@ class AlexaEntity:
yield from interface.serialize_properties()
def serialize_discovery(self):
def serialize_discovery(self) -> dict[str, Any]:
"""Serialize the entity for discovery."""
result = {
result: dict[str, Any] = {
"displayCategories": self.display_categories(),
"cookie": {},
"endpointId": self.alexa_id(),
@ -366,7 +366,7 @@ def async_get_entities(
hass: HomeAssistant, config: AbstractConfig
) -> list[AlexaEntity]:
"""Return all entities that are supported by Alexa."""
entities = []
entities: list[AlexaEntity] = []
for state in hass.states.async_all():
if state.entity_id in CLOUD_NEVER_EXPOSED_ENTITIES:
continue

View File

@ -3,8 +3,10 @@ import enum
import logging
from typing import Any
from aiohttp.web import Response
from homeassistant.components import http
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import intent
from homeassistant.util.decorator import Registry
@ -18,7 +20,7 @@ HANDLERS = Registry() # type: ignore[var-annotated]
INTENTS_API_ENDPOINT = "/api/alexa"
class SpeechType(enum.Enum):
class SpeechType(enum.StrEnum):
"""The Alexa speech types."""
plaintext = "PlainText"
@ -28,7 +30,7 @@ class SpeechType(enum.Enum):
SPEECH_MAPPINGS = {"plain": SpeechType.plaintext, "ssml": SpeechType.ssml}
class CardType(enum.Enum):
class CardType(enum.StrEnum):
"""The Alexa card types."""
simple = "Simple"
@ -36,12 +38,12 @@ class CardType(enum.Enum):
@callback
def async_setup(hass):
def async_setup(hass: HomeAssistant) -> None:
"""Activate Alexa component."""
hass.http.register_view(AlexaIntentsView)
async def async_setup_intents(hass):
async def async_setup_intents(hass: HomeAssistant) -> None:
"""Do intents setup.
Right now this module does not expose any, but the intent component breaks
@ -60,15 +62,15 @@ class AlexaIntentsView(http.HomeAssistantView):
url = INTENTS_API_ENDPOINT
name = "api:alexa"
async def post(self, request):
async def post(self, request: http.HomeAssistantRequest) -> Response | bytes:
"""Handle Alexa."""
hass = request.app["hass"]
message = await request.json()
hass: HomeAssistant = request.app["hass"]
message: dict[str, Any] = await request.json()
_LOGGER.debug("Received Alexa request: %s", message)
try:
response = await async_handle_message(hass, message)
response: dict[str, Any] = await async_handle_message(hass, message)
return b"" if response is None else self.json(response)
except UnknownRequest as err:
_LOGGER.warning(str(err))
@ -99,15 +101,19 @@ class AlexaIntentsView(http.HomeAssistantView):
)
def intent_error_response(hass, message, error):
def intent_error_response(
hass: HomeAssistant, message: dict[str, Any], error: str
) -> dict[str, Any]:
"""Return an Alexa response that will speak the error message."""
alexa_intent_info = message.get("request").get("intent")
alexa_response = AlexaResponse(hass, alexa_intent_info)
alexa_intent_info = message["request"].get("intent")
alexa_response = AlexaIntentResponse(hass, alexa_intent_info)
alexa_response.add_speech(SpeechType.plaintext, error)
return alexa_response.as_dict()
async def async_handle_message(hass, message):
async def async_handle_message(
hass: HomeAssistant, message: dict[str, Any]
) -> dict[str, Any]:
"""Handle an Alexa intent.
Raises:
@ -117,7 +123,7 @@ async def async_handle_message(hass, message):
- intent.IntentError
"""
req = message.get("request")
req = message["request"]
req_type = req["type"]
if not (handler := HANDLERS.get(req_type)):
@ -129,7 +135,9 @@ async def async_handle_message(hass, message):
@HANDLERS.register("SessionEndedRequest")
@HANDLERS.register("IntentRequest")
@HANDLERS.register("LaunchRequest")
async def async_handle_intent(hass, message):
async def async_handle_intent(
hass: HomeAssistant, message: dict[str, Any]
) -> dict[str, Any]:
"""Handle an intent request.
Raises:
@ -138,9 +146,9 @@ async def async_handle_intent(hass, message):
- intent.IntentError
"""
req = message.get("request")
req = message["request"]
alexa_intent_info = req.get("intent")
alexa_response = AlexaResponse(hass, alexa_intent_info)
alexa_response = AlexaIntentResponse(hass, alexa_intent_info)
if req["type"] == "LaunchRequest":
intent_name = (
@ -187,7 +195,7 @@ def resolve_slot_data(key: str, request: dict[str, Any]) -> dict[str, str]:
# passes the id and name of the nearest possible slot resolution. For
# reference to the request object structure, see the Alexa docs:
# https://tinyurl.com/ybvm7jhs
resolved_data = {}
resolved_data: dict[str, Any] = {}
resolved_data["value"] = request["value"]
resolved_data["id"] = ""
@ -226,18 +234,18 @@ def resolve_slot_data(key: str, request: dict[str, Any]) -> dict[str, str]:
return resolved_data
class AlexaResponse:
class AlexaIntentResponse:
"""Help generating the response for Alexa."""
def __init__(self, hass, intent_info):
def __init__(self, hass: HomeAssistant, intent_info: dict[str, Any] | None) -> None:
"""Initialize the response."""
self.hass = hass
self.speech = None
self.card = None
self.reprompt = None
self.session_attributes = {}
self.speech: dict[str, Any] | None = None
self.card: dict[str, Any] | None = None
self.reprompt: dict[str, Any] | None = None
self.session_attributes: dict[str, Any] = {}
self.should_end_session = True
self.variables = {}
self.variables: dict[str, Any] = {}
# Intent is None if request was a LaunchRequest or SessionEndedRequest
if intent_info is not None:
@ -252,7 +260,7 @@ class AlexaResponse:
self.variables[_key] = _slot_data["value"]
self.variables[_key + "_Id"] = _slot_data["id"]
def add_card(self, card_type, title, content):
def add_card(self, card_type: CardType, title: str, content: str) -> None:
"""Add a card to the response."""
assert self.card is None
@ -266,7 +274,7 @@ class AlexaResponse:
card["content"] = content
self.card = card
def add_speech(self, speech_type, text):
def add_speech(self, speech_type: SpeechType, text: str) -> None:
"""Add speech to the response."""
assert self.speech is None
@ -274,7 +282,7 @@ class AlexaResponse:
self.speech = {"type": speech_type.value, key: text}
def add_reprompt(self, speech_type, text):
def add_reprompt(self, speech_type: SpeechType, text) -> None:
"""Add reprompt if user does not answer."""
assert self.reprompt is None
@ -284,9 +292,9 @@ class AlexaResponse:
self.reprompt = {"type": speech_type.value, key: text}
def as_dict(self):
def as_dict(self) -> dict[str, Any]:
"""Return response in an Alexa valid dict."""
response = {"shouldEndSession": self.should_end_session}
response: dict[str, Any] = {"shouldEndSession": self.should_end_session}
if self.card is not None:
response["card"] = self.card

View File

@ -1,6 +1,9 @@
"""Alexa Resources and Assets."""
from typing import Any
class AlexaGlobalCatalog:
"""The Global Alexa catalog.
@ -207,36 +210,40 @@ class AlexaCapabilityResource:
https://developer.amazon.com/docs/device-apis/resources-and-assets.html#capability-resources
"""
def __init__(self, labels):
def __init__(self, labels: list[str]) -> None:
"""Initialize an Alexa resource."""
self._resource_labels = []
for label in labels:
self._resource_labels.append(label)
def serialize_capability_resources(self):
def serialize_capability_resources(self) -> dict[str, list[dict[str, Any]]]:
"""Return capabilityResources object serialized for an API response."""
return self.serialize_labels(self._resource_labels)
def serialize_configuration(self):
def serialize_configuration(self) -> dict[str, Any]:
"""Return serialized configuration for an API response.
Return ModeResources, PresetResources friendlyNames serialized.
"""
return []
raise NotImplementedError()
def serialize_labels(self, resources):
def serialize_labels(self, resources: list[str]) -> dict[str, list[dict[str, Any]]]:
"""Return serialized labels for an API response.
Returns resource label objects for friendlyNames serialized.
"""
labels = []
labels: list[dict[str, Any]] = []
label_dict: dict[str, Any]
for label in resources:
if label in AlexaGlobalCatalog.__dict__.values():
label = {"@type": "asset", "value": {"assetId": label}}
label_dict = {"@type": "asset", "value": {"assetId": label}}
else:
label = {"@type": "text", "value": {"text": label, "locale": "en-US"}}
label_dict = {
"@type": "text",
"value": {"text": label, "locale": "en-US"},
}
labels.append(label)
labels.append(label_dict)
return {"friendlyNames": labels}
@ -247,22 +254,22 @@ class AlexaModeResource(AlexaCapabilityResource):
https://developer.amazon.com/docs/device-apis/resources-and-assets.html#capability-resources
"""
def __init__(self, labels, ordered=False):
def __init__(self, labels: list[str], ordered: bool = False) -> None:
"""Initialize an Alexa modeResource."""
super().__init__(labels)
self._supported_modes = []
self._mode_ordered = ordered
self._supported_modes: list[dict[str, Any]] = []
self._mode_ordered: bool = ordered
def add_mode(self, value, labels):
def add_mode(self, value: str, labels: list[str]) -> None:
"""Add mode to the supportedModes object."""
self._supported_modes.append({"value": value, "labels": labels})
def serialize_configuration(self):
def serialize_configuration(self) -> dict[str, Any]:
"""Return serialized configuration for an API response.
Returns configuration for ModeResources friendlyNames serialized.
"""
mode_resources = []
mode_resources: list[dict[str, Any]] = []
for mode in self._supported_modes:
result = {
"value": mode["value"],
@ -282,10 +289,17 @@ class AlexaPresetResource(AlexaCapabilityResource):
https://developer.amazon.com/docs/device-apis/resources-and-assets.html#presetresources
"""
def __init__(self, labels, min_value, max_value, precision, unit=None):
def __init__(
self,
labels: list[str],
min_value: int | float,
max_value: int | float,
precision: int | float,
unit: str | None = None,
) -> None:
"""Initialize an Alexa presetResource."""
super().__init__(labels)
self._presets = []
self._presets: list[dict[str, Any]] = []
self._minimum_value = min_value
self._maximum_value = max_value
self._precision = precision
@ -293,16 +307,16 @@ class AlexaPresetResource(AlexaCapabilityResource):
if unit in AlexaGlobalCatalog.__dict__.values():
self._unit_of_measure = unit
def add_preset(self, value, labels):
def add_preset(self, value: int | float, labels: list[str]) -> None:
"""Add preset to configuration presets array."""
self._presets.append({"value": value, "labels": labels})
def serialize_configuration(self):
def serialize_configuration(self) -> dict[str, Any]:
"""Return serialized configuration for an API response.
Returns configuration for PresetResources friendlyNames serialized.
"""
configuration = {
configuration: dict[str, Any] = {
"supportedRange": {
"minimumValue": self._minimum_value,
"maximumValue": self._maximum_value,
@ -372,26 +386,28 @@ class AlexaSemantics:
DIRECTIVE_MODE_SET_MODE = "SetMode"
DIRECTIVE_MODE_ADJUST_MODE = "AdjustMode"
def __init__(self):
def __init__(self) -> None:
"""Initialize an Alexa modeResource."""
self._action_mappings = []
self._state_mappings = []
self._action_mappings: list[dict[str, Any]] = []
self._state_mappings: list[dict[str, Any]] = []
def _add_action_mapping(self, semantics):
def _add_action_mapping(self, semantics: dict[str, Any]) -> None:
"""Add action mapping between actions and interface directives."""
self._action_mappings.append(semantics)
def _add_state_mapping(self, semantics):
def _add_state_mapping(self, semantics: dict[str, Any]) -> None:
"""Add state mapping between states and interface directives."""
self._state_mappings.append(semantics)
def add_states_to_value(self, states, value):
def add_states_to_value(self, states: list[str], value: int | float) -> None:
"""Add StatesToValue stateMappings."""
self._add_state_mapping(
{"@type": self.STATES_TO_VALUE, "states": states, "value": value}
)
def add_states_to_range(self, states, min_value, max_value):
def add_states_to_range(
self, states: list[str], min_value: int | float, max_value: int | float
) -> None:
"""Add StatesToRange stateMappings."""
self._add_state_mapping(
{
@ -401,7 +417,9 @@ class AlexaSemantics:
}
)
def add_action_to_directive(self, actions, directive, payload):
def add_action_to_directive(
self, actions: list[str], directive: str, payload: dict[str, Any]
) -> None:
"""Add ActionsToDirective actionMappings."""
self._add_action_mapping(
{
@ -411,9 +429,9 @@ class AlexaSemantics:
}
)
def serialize_semantics(self):
def serialize_semantics(self) -> dict[str, Any]:
"""Return semantics object serialized for an API response."""
semantics = {}
semantics: dict[str, Any] = {}
if self._action_mappings:
semantics[self.MAPPINGS_ACTION] = self._action_mappings
if self._state_mappings: