Adjust device_automation type hints in core components (#72207)

pull/72126/head
epenet 2022-05-23 16:07:34 +02:00 committed by GitHub
parent fb53e39f05
commit 5cfb31d28a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 176 additions and 51 deletions

View File

@ -1,8 +1,6 @@
"""Device automation helpers for entity."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant.components.automation import (
@ -91,7 +89,7 @@ async def _async_get_automations(
async def async_get_triggers(
hass: HomeAssistant, device_id: str, domain: str
) -> list[dict[str, Any]]:
) -> list[dict[str, str]]:
"""List device triggers."""
return await _async_get_automations(hass, device_id, ENTITY_TRIGGERS, domain)

View File

@ -1,8 +1,6 @@
"""Device automation helpers for toggle entity."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant.components.automation import (
@ -228,7 +226,7 @@ async def async_get_conditions(
async def async_get_triggers(
hass: HomeAssistant, device_id: str, domain: str
) -> list[dict[str, Any]]:
) -> list[dict[str, str]]:
"""List device triggers."""
triggers = await entity.async_get_triggers(hass, device_id, domain)
triggers.extend(

View File

@ -3,11 +3,16 @@ import logging
import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import CONF_EVENT, CONF_PLATFORM, CONF_SOURCE, CONF_ZONE
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import condition, config_validation as cv
from homeassistant.helpers.config_validation import entity_domain
from homeassistant.helpers.event import TrackStates, async_track_state_change_filtered
from homeassistant.helpers.typing import ConfigType
from . import DOMAIN
@ -36,10 +41,15 @@ def source_match(state, source):
return state and state.attributes.get("source") == source
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
trigger_data = automation_info["trigger_data"]
source = config.get(CONF_SOURCE).lower()
source: str = config[CONF_SOURCE].lower()
zone_entity_id = config.get(CONF_ZONE)
trigger_event = config.get(CONF_EVENT)
job = HassJob(action)

View File

@ -1,14 +1,25 @@
"""Home Assistant trigger dispatcher."""
import importlib
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.components.device_automation.trigger import (
DeviceAutomationTriggerProtocol,
)
from homeassistant.const import CONF_PLATFORM
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.typing import ConfigType
def _get_trigger_platform(config):
def _get_trigger_platform(config: ConfigType) -> DeviceAutomationTriggerProtocol:
return importlib.import_module(f"..triggers.{config[CONF_PLATFORM]}", __name__)
async def async_validate_trigger_config(hass, config):
async def async_validate_trigger_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
platform = _get_trigger_platform(config)
if hasattr(platform, "async_validate_trigger_config"):
@ -17,7 +28,12 @@ async def async_validate_trigger_config(hass, config):
return platform.TRIGGER_SCHEMA(config)
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Attach trigger of specified platform."""
platform = _get_trigger_platform(config)
return await platform.async_attach_trigger(hass, config, action, automation_info)

View File

@ -1,9 +1,14 @@
"""Offer Home Assistant core automation rules."""
import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import CONF_EVENT, CONF_PLATFORM, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
# mypy: allow-untyped-defs
@ -18,7 +23,12 @@ TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
)
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for events based on configuration."""
trigger_data = automation_info["trigger_data"]
event = config.get(CONF_EVENT)

View File

@ -4,6 +4,10 @@ import logging
import voluptuous as vol
from homeassistant import exceptions
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import (
CONF_ABOVE,
CONF_ATTRIBUTE,
@ -81,10 +85,15 @@ async def async_validate_trigger_config(
async def async_attach_trigger(
hass, config, action, automation_info, *, platform_type="numeric_state"
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
*,
platform_type: str = "numeric_state",
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
entity_ids = config.get(CONF_ENTITY_ID)
entity_ids: list[str] = config[CONF_ENTITY_ID]
below = config.get(CONF_BELOW)
above = config.get(CONF_ABOVE)
time_delta = config.get(CONF_FOR)

View File

@ -7,6 +7,10 @@ import logging
import voluptuous as vol
from homeassistant import exceptions
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import CONF_ATTRIBUTE, CONF_FOR, CONF_PLATFORM, MATCH_ALL
from homeassistant.core import (
CALLBACK_TYPE,
@ -92,9 +96,9 @@ async def async_validate_trigger_config(
async def async_attach_trigger(
hass: HomeAssistant,
config,
action,
automation_info,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
*,
platform_type: str = "state",
) -> CALLBACK_TYPE:

View File

@ -5,6 +5,10 @@ from functools import partial
import voluptuous as vol
from homeassistant.components import sensor
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
CONF_AT,
@ -12,13 +16,14 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.event import (
async_track_point_in_time,
async_track_state_change_event,
async_track_time_change,
)
from homeassistant.helpers.typing import ConfigType
import homeassistant.util.dt as dt_util
# mypy: allow-untyped-defs, no-check-untyped-defs
@ -37,10 +42,15 @@ TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
)
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
trigger_data = automation_info["trigger_data"]
entities = {}
entities: dict[str, CALLBACK_TYPE] = {}
removes = []
job = HassJob(action)

View File

@ -1,10 +1,15 @@
"""Offer time listening automation rules."""
import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import CONF_PLATFORM
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.event import async_track_time_change
from homeassistant.helpers.typing import ConfigType
# mypy: allow-untyped-defs, no-check-untyped-defs
@ -55,7 +60,12 @@ TRIGGER_SCHEMA = vol.All(
)
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
trigger_data = automation_info["trigger_data"]
hours = config.get(CONF_HOURS)

View File

@ -3,7 +3,7 @@ from __future__ import annotations
from collections.abc import Callable
import logging
from typing import Any, cast
from typing import cast
import attr
import voluptuous as vol
@ -290,9 +290,9 @@ async def async_removed_from_device(hass: HomeAssistant, device_id: str) -> None
async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, Any]]:
) -> list[dict[str, str]]:
"""List device triggers for MQTT devices."""
triggers: list[dict] = []
triggers: list[dict[str, str]] = []
if DEVICE_TRIGGERS not in hass.data:
return triggers

View File

@ -5,9 +5,14 @@ import logging
import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import CONF_PAYLOAD, CONF_PLATFORM, CONF_VALUE_TEMPLATE
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, template
from homeassistant.helpers.typing import ConfigType
from .. import mqtt
from .const import CONF_ENCODING, CONF_QOS, CONF_TOPIC, DEFAULT_ENCODING, DEFAULT_QOS
@ -31,7 +36,12 @@ TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
_LOGGER = logging.getLogger(__name__)
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
trigger_data = automation_info["trigger_data"]
topic = config[CONF_TOPIC]

View File

@ -3,15 +3,20 @@ from datetime import timedelta
import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import (
CONF_EVENT,
CONF_OFFSET,
CONF_PLATFORM,
SUN_EVENT_SUNRISE,
)
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import async_track_sunrise, async_track_sunset
from homeassistant.helpers.typing import ConfigType
# mypy: allow-untyped-defs, no-check-untyped-defs
@ -24,7 +29,12 @@ TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
)
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for events based on configuration."""
trigger_data = automation_info["trigger_data"]
event = config.get(CONF_EVENT)

View File

@ -4,15 +4,20 @@ import logging
import voluptuous as vol
from homeassistant import exceptions
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import CONF_FOR, CONF_PLATFORM, CONF_VALUE_TEMPLATE
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, template
from homeassistant.helpers.event import (
TrackTemplate,
async_call_later,
async_track_template_result,
)
from homeassistant.helpers.template import result_as_boolean
from homeassistant.helpers.template import Template, result_as_boolean
from homeassistant.helpers.typing import ConfigType
# mypy: allow-untyped-defs, no-check-untyped-defs
@ -28,11 +33,16 @@ TRIGGER_SCHEMA = IF_ACTION_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
async def async_attach_trigger(
hass, config, action, automation_info, *, platform_type="template"
):
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
*,
platform_type: str = "template",
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
trigger_data = automation_info["trigger_data"]
value_template = config.get(CONF_VALUE_TEMPLATE)
value_template: Template = config[CONF_VALUE_TEMPLATE]
value_template.hass = hass
time_delta = config.get(CONF_FOR)
template.attach(hass, time_delta)

View File

@ -4,9 +4,14 @@ from functools import partial
from aiohttp import hdrs
import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import CONF_PLATFORM, CONF_WEBHOOK_ID
from homeassistant.core import HassJob, callback
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from . import async_register, async_unregister
@ -37,10 +42,15 @@ async def _handle_webhook(job, trigger_data, hass, webhook_id, request):
hass.async_run_hass_job(job, {"trigger": result})
async def async_attach_trigger(hass, config, action, automation_info):
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Trigger based on incoming webhooks."""
trigger_data = automation_info["trigger_data"]
webhook_id = config.get(CONF_WEBHOOK_ID)
webhook_id: str = config[CONF_WEBHOOK_ID]
job = HassJob(action)
async_register(
hass,

View File

@ -3,6 +3,10 @@ import logging
import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.const import (
ATTR_FRIENDLY_NAME,
CONF_ENTITY_ID,
@ -56,11 +60,16 @@ async def async_validate_trigger_config(
async def async_attach_trigger(
hass, config, action, automation_info, *, platform_type: str = "zone"
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
*,
platform_type: str = "zone",
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
trigger_data = automation_info["trigger_data"]
entity_id = config.get(CONF_ENTITY_ID)
entity_id: list[str] = config[CONF_ENTITY_ID]
zone_entity_id = config.get(CONF_ZONE)
event = config.get(CONF_EVENT)
job = HassJob(action)

View File

@ -5,7 +5,7 @@ import asyncio
from collections.abc import Callable
import functools
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
import voluptuous as vol
@ -16,13 +16,20 @@ from homeassistant.loader import IntegrationNotFound, async_get_integration
from .typing import ConfigType, TemplateVarsType
if TYPE_CHECKING:
from homeassistant.components.device_automation.trigger import (
DeviceAutomationTriggerProtocol,
)
_PLATFORM_ALIASES = {
"device_automation": ("device",),
"homeassistant": ("event", "numeric_state", "state", "time_pattern", "time"),
}
async def _async_get_trigger_platform(hass: HomeAssistant, config: ConfigType) -> Any:
async def _async_get_trigger_platform(
hass: HomeAssistant, config: ConfigType
) -> DeviceAutomationTriggerProtocol:
platform_and_sub_type = config[CONF_PLATFORM].split(".")
platform = platform_and_sub_type[0]
for alias, triggers in _PLATFORM_ALIASES.items():
@ -86,6 +93,10 @@ async def async_initialize_triggers(
variables: TemplateVarsType = None,
) -> CALLBACK_TYPE | None:
"""Initialize triggers."""
from homeassistant.components.automation import ( # pylint:disable=[import-outside-toplevel]
AutomationTriggerData,
AutomationTriggerInfo,
)
triggers = []
for idx, conf in enumerate(trigger_config):
@ -96,14 +107,14 @@ async def async_initialize_triggers(
platform = await _async_get_trigger_platform(hass, conf)
trigger_id = conf.get(CONF_ID, f"{idx}")
trigger_idx = f"{idx}"
trigger_data = {"id": trigger_id, "idx": trigger_idx}
info = {
"domain": domain,
"name": name,
"home_assistant_start": home_assistant_start,
"variables": variables,
"trigger_data": trigger_data,
}
trigger_data = AutomationTriggerData(id=trigger_id, idx=trigger_idx)
info = AutomationTriggerInfo(
domain=domain,
name=name,
home_assistant_start=home_assistant_start,
variables=variables,
trigger_data=trigger_data,
)
triggers.append(
platform.async_attach_trigger(