hass-core/homeassistant/components/zha/device_action.py
Alexei Chetroi 3385893b77
ZHA device channel refactoring (#31971)
* Add ZHA core typing helper.
* Add aux_channels to ZHA rule matching.
* Add match rule claim_channels() method.
* Expose underlying zigpy device.
* Not sure we need this one.
* Move "base" channels.
* Framework for channel discovery.
* Make DEVICE_CLASS and REMOTE_DEVICE_TYPE default dicts.
* Remove attribute reporting configuration registry.
* Refactor channels.
- Refactor zha events
- Use compound IDs and unique_ids
- Refactor signal dispatching on attribute updates

* Use unique id compatible with entities unique ids.
* Refactor ZHA Entity registry.
Let match rule to check for the match.

* Refactor discovery to use new channels.
* Cleanup ZDO channel.
Remove unused zha store call.

* Handle channel configuration and initialization.
* Refactor ZHA Device to use new channels.
* Refactor ZHA Gateway to use new discovery framework.
Use hass.data for entity info intermediate store.

* Don't keep entities in hass.data.
* ZHA gateway new discovery framework.
* Refactor ZHA platform loading.
* Don't update ZHA entities, when restoring from zigpy.
* ZHA entity discover tests.
* Add AnalogInput sensor.
* Remove 0xFC02 based entity from Keen smart vents.
* Clean up IAS channels.
* Refactor entity restoration.
* Fix lumi.router entities name.
* Rename EndpointsChannel to ChannelPool.
* Make Channels.pools a list.
* Fix cover test.
* Fix FakeDevice class.
* Fix device actions.
* Fix channels typing.
* Revert update_before_add=False
* Refactor channel class matching.
* Use a helper function for adding entities.
* Make Pylint happy.
* Rebase cleanup.
* Update coverage for ZHA device type overrides.
* Use cluster_id for single output cluster registry.
* Remove ZHA typing from coverage.
* Fix tests.
* Address comments.
* Address comments.
2020-02-21 18:06:57 -05:00

93 lines
2.6 KiB
Python

"""Provides device actions for ZHA devices."""
from typing import List
import voluptuous as vol
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_TYPE
from homeassistant.core import Context, HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from . import DOMAIN
from .api import SERVICE_WARNING_DEVICE_SQUAWK, SERVICE_WARNING_DEVICE_WARN
from .core.const import CHANNEL_IAS_WD
from .core.helpers import async_get_zha_device
ACTION_SQUAWK = "squawk"
ACTION_WARN = "warn"
ATTR_DATA = "data"
ATTR_IEEE = "ieee"
CONF_ZHA_ACTION_TYPE = "zha_action_type"
ZHA_ACTION_TYPE_SERVICE_CALL = "service_call"
ACTION_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
{vol.Required(CONF_DOMAIN): DOMAIN, vol.Required(CONF_TYPE): str}
)
DEVICE_ACTIONS = {
CHANNEL_IAS_WD: [
{CONF_TYPE: ACTION_SQUAWK, CONF_DOMAIN: DOMAIN},
{CONF_TYPE: ACTION_WARN, CONF_DOMAIN: DOMAIN},
]
}
DEVICE_ACTION_TYPES = {
ACTION_SQUAWK: ZHA_ACTION_TYPE_SERVICE_CALL,
ACTION_WARN: ZHA_ACTION_TYPE_SERVICE_CALL,
}
SERVICE_NAMES = {
ACTION_SQUAWK: SERVICE_WARNING_DEVICE_SQUAWK,
ACTION_WARN: SERVICE_WARNING_DEVICE_WARN,
}
async def async_call_action_from_config(
hass: HomeAssistant,
config: ConfigType,
variables: TemplateVarsType,
context: Context,
) -> None:
"""Perform an action based on configuration."""
await ZHA_ACTION_TYPES[DEVICE_ACTION_TYPES[config[CONF_TYPE]]](
hass, config, variables, context
)
async def async_get_actions(hass: HomeAssistant, device_id: str) -> List[dict]:
"""List device actions."""
zha_device = await async_get_zha_device(hass, device_id)
cluster_channels = [
ch.name
for pool in zha_device.channels.pools
for ch in pool.claimed_channels.values()
]
actions = [
action
for channel in DEVICE_ACTIONS
for action in DEVICE_ACTIONS[channel]
if channel in cluster_channels
]
for action in actions:
action[CONF_DEVICE_ID] = device_id
return actions
async def _execute_service_based_action(
hass: HomeAssistant,
config: ACTION_SCHEMA,
variables: TemplateVarsType,
context: Context,
) -> None:
action_type = config[CONF_TYPE]
service_name = SERVICE_NAMES[action_type]
zha_device = await async_get_zha_device(hass, config[CONF_DEVICE_ID])
service_data = {ATTR_IEEE: str(zha_device.ieee)}
await hass.services.async_call(
DOMAIN, service_name, service_data, blocking=True, context=context
)
ZHA_ACTION_TYPES = {ZHA_ACTION_TYPE_SERVICE_CALL: _execute_service_based_action}