evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add support for Pico and Shade remotes to Lutron Caseta (#45315)

Browse source
This commit is contained in:
J. Nick Koston 2021-01-26 16:32:08 -06:00 • committed by GitHub
parent b533b91b10
commit 25f411ef6e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 982 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

190
homeassistant/components/lutron_caseta/__init__.py
View file

@ -2,19 +2,44 @@
import asyncio
import logging
from aiolip import LIP
from aiolip.data import LIPMode
from aiolip.protocol import LIP_BUTTON_PRESS
from pylutron_caseta.smartbridge import Smartbridge
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST
from homeassistant.core import callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from .const import CONF_CA_CERTS, CONF_CERTFILE, CONF_KEYFILE
from .const import (
ACTION_PRESS,
ACTION_RELEASE,
ATTR_ACTION,
ATTR_AREA_NAME,
ATTR_BUTTON_NUMBER,
ATTR_DEVICE_NAME,
ATTR_SERIAL,
ATTR_TYPE,
BRIDGE_DEVICE,
BRIDGE_DEVICE_ID,
BRIDGE_LEAP,
BRIDGE_LIP,
BUTTON_DEVICES,
CONF_CA_CERTS,
CONF_CERTFILE,
CONF_KEYFILE,
DOMAIN,
LUTRON_CASETA_BUTTON_EVENT,
MANUFACTURER,
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = "lutron_caseta"
DATA_BRIDGE_CONFIG = "lutron_caseta_bridges"
CONFIG_SCHEMA = vol.Schema(
@ -76,14 +101,29 @@ async def async_setup_entry(hass, config_entry):
await bridge.connect()
if not bridge.is_connected():
await bridge.close()
_LOGGER.error("Unable to connect to Lutron Caseta bridge at %s", host)
return False
raise ConfigEntryNotReady
_LOGGER.debug("Connected to Lutron Caseta bridge at %s", host)
_LOGGER.debug("Connected to Lutron Caseta bridge via LEAP at %s", host)
devices = bridge.get_devices()
bridge_device = devices[BRIDGE_DEVICE_ID]
await _async_register_bridge_device(hass, config_entry.entry_id, bridge_device)
# Store this bridge (keyed by entry_id) so it can be retrieved by the
# components we're setting up.
hass.data[DOMAIN][config_entry.entry_id] = bridge
hass.data[DOMAIN][config_entry.entry_id] = {
BRIDGE_LEAP: bridge,
BRIDGE_DEVICE: bridge_device,
BUTTON_DEVICES: {},
BRIDGE_LIP: None,
}
if bridge.lip_devices:
# If the bridge also supports LIP (Lutron Integration Protocol)
# we can fire events when pico buttons are pressed to allow
# pico remotes to control other devices.
await async_setup_lip(hass, config_entry, bridge.lip_devices)
for component in LUTRON_CASETA_COMPONENTS:
hass.async_create_task(
@ -93,10 +133,139 @@ async def async_setup_entry(hass, config_entry):
return True
async def async_setup_lip(hass, config_entry, lip_devices):
"""Connect to the bridge via Lutron Integration Protocol to watch for pico remotes."""
host = config_entry.data[CONF_HOST]
config_entry_id = config_entry.entry_id
data = hass.data[DOMAIN][config_entry_id]
bridge_device = data[BRIDGE_DEVICE]
bridge = data[BRIDGE_LEAP]
lip = LIP()
try:
await lip.async_connect(host)
except asyncio.TimeoutError:
_LOGGER.error("Failed to connect to via LIP at %s:23", host)
return
_LOGGER.debug("Connected to Lutron Caseta bridge via LIP at %s:23", host)
button_devices_by_lip_id = _async_merge_lip_leap_data(lip_devices, bridge)
button_devices_by_dr_id = await _async_register_button_devices(
hass, config_entry_id, bridge_device, button_devices_by_lip_id
)
_async_subscribe_pico_remote_events(hass, lip, button_devices_by_lip_id)
data[BUTTON_DEVICES] = button_devices_by_dr_id
data[BRIDGE_LIP] = lip
@callback
def _async_merge_lip_leap_data(lip_devices, bridge):
"""Merge the leap data into the lip data."""
sensor_devices = bridge.get_devices_by_domain("sensor")
button_devices_by_id = {
id: device for id, device in lip_devices.items() if "Buttons" in device
}
sensor_devices_by_name = {device["name"]: device for device in sensor_devices}
# Add the leap data into the lip data
# so we know the type, model, and serial
for device in button_devices_by_id.values():
area = device.get("Area", {}).get("Name", "")
name = device["Name"]
leap_name = f"{area}_{name}"
device["leap_name"] = leap_name
leap_device_data = sensor_devices_by_name.get(leap_name)
if leap_device_data is None:
continue
for key in ("type", "model", "serial"):
val = leap_device_data.get(key)
if val is not None:
device[key] = val
_LOGGER.debug("Button Devices: %s", button_devices_by_id)
return button_devices_by_id
async def _async_register_bridge_device(hass, config_entry_id, bridge_device):
"""Register the bridge device in the device registry."""
device_registry = await dr.async_get_registry(hass)
device_registry.async_get_or_create(
name=bridge_device["name"],
manufacturer=MANUFACTURER,
config_entry_id=config_entry_id,
identifiers={(DOMAIN, bridge_device["serial"])},
model=f"{bridge_device['model']} ({bridge_device['type']})",
)
async def _async_register_button_devices(
hass, config_entry_id, bridge_device, button_devices_by_id
):
"""Register button devices (Pico Remotes) in the device registry."""
device_registry = await dr.async_get_registry(hass)
button_devices_by_dr_id = {}
for device in button_devices_by_id.values():
if "serial" not in device:
continue
dr_device = device_registry.async_get_or_create(
name=device["leap_name"],
manufacturer=MANUFACTURER,
config_entry_id=config_entry_id,
identifiers={(DOMAIN, device["serial"])},
model=f"{device['model']} ({device['type']})",
via_device=(DOMAIN, bridge_device["serial"]),
)
button_devices_by_dr_id[dr_device.id] = device
return button_devices_by_dr_id
@callback
def _async_subscribe_pico_remote_events(hass, lip, button_devices_by_id):
"""Subscribe to lutron events."""
@callback
def _async_lip_event(lip_message):
if lip_message.mode != LIPMode.DEVICE:
return
device = button_devices_by_id.get(lip_message.integration_id)
if not device:
return
if lip_message.value == LIP_BUTTON_PRESS:
action = ACTION_PRESS
else:
action = ACTION_RELEASE
hass.bus.async_fire(
LUTRON_CASETA_BUTTON_EVENT,
{
ATTR_SERIAL: device.get("serial"),
ATTR_TYPE: device.get("type"),
ATTR_BUTTON_NUMBER: lip_message.action_number,
ATTR_DEVICE_NAME: device["Name"],
ATTR_AREA_NAME: device.get("Area", {}).get("Name"),
ATTR_ACTION: action,
},
)
lip.subscribe(_async_lip_event)
asyncio.create_task(lip.async_run())
async def async_unload_entry(hass, config_entry):
"""Unload the bridge bridge from a config entry."""
hass.data[DOMAIN][config_entry.entry_id].close()
data = hass.data[DOMAIN][config_entry.entry_id]
data[BRIDGE_LEAP].close()
if data[BRIDGE_LIP]:
await data[BRIDGE_LIP].async_stop()
unload_ok = all(
await asyncio.gather(
@ -116,14 +285,16 @@ async def async_unload_entry(hass, config_entry):
class LutronCasetaDevice(Entity):
"""Common base class for all Lutron Caseta devices."""
def __init__(self, device, bridge):
def __init__(self, device, bridge, bridge_device):
"""Set up the base class.
[:param]device the device metadata
[:param]bridge the smartbridge object
[:param]bridge_device a dict with the details of the bridge
"""
self._device = device
self._smartbridge = bridge
self._bridge_device = bridge_device
async def async_added_to_hass(self):
"""Register callbacks."""
@ -155,8 +326,9 @@ class LutronCasetaDevice(Entity):
return {
"identifiers": {(DOMAIN, self.serial)},
"name": self.name,
"manufacturer": "Lutron",
"model": self._device["model"],
"manufacturer": MANUFACTURER,
"model": f"{self._device['model']} ({self._device['type']})",
"via_device": (DOMAIN, self._bridge_device["serial"]),
}
@property

7
homeassistant/components/lutron_caseta/binary_sensor.py
View file

@ -7,6 +7,7 @@ from homeassistant.components.binary_sensor import (
)
from . import DOMAIN as CASETA_DOMAIN, LutronCasetaDevice
from .const import BRIDGE_DEVICE, BRIDGE_LEAP
async def async_setup_entry(hass, config_entry, async_add_entities):
@ -17,11 +18,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"""
entities = []
bridge = hass.data[CASETA_DOMAIN][config_entry.entry_id]
data = hass.data[CASETA_DOMAIN][config_entry.entry_id]
bridge = data[BRIDGE_LEAP]
bridge_device = data[BRIDGE_DEVICE]
occupancy_groups = bridge.occupancy_groups
for occupancy_group in occupancy_groups.values():
entity = LutronOccupancySensor(occupancy_group, bridge)
entity = LutronOccupancySensor(occupancy_group, bridge, bridge_device)
entities.append(entity)
async_add_entities(entities, True)

2
homeassistant/components/lutron_caseta/config_flow.py
View file

@ -12,7 +12,6 @@ from homeassistant.components.zeroconf import ATTR_HOSTNAME
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import callback
from . import DOMAIN # pylint: disable=unused-import
from .const import (
ABORT_REASON_ALREADY_CONFIGURED,
ABORT_REASON_CANNOT_CONNECT,
@ -22,6 +21,7 @@ from .const import (
ERROR_CANNOT_CONNECT,
STEP_IMPORT_FAILED,
)
from .const import DOMAIN # pylint: disable=unused-import
HOSTNAME = "hostname"

25
homeassistant/components/lutron_caseta/const.py
View file

@ -1,5 +1,7 @@
"""Lutron Caseta constants."""
DOMAIN = "lutron_caseta"
CONF_KEYFILE = "keyfile"
CONF_CERTFILE = "certfile"
CONF_CA_CERTS = "ca_certs"
@ -8,3 +10,26 @@ STEP_IMPORT_FAILED = "import_failed"
ERROR_CANNOT_CONNECT = "cannot_connect"
ABORT_REASON_CANNOT_CONNECT = "cannot_connect"
ABORT_REASON_ALREADY_CONFIGURED = "already_configured"
BRIDGE_LEAP = "leap"
BRIDGE_LIP = "lip"
BRIDGE_DEVICE = "bridge_device"
BUTTON_DEVICES = "button_devices"
LUTRON_CASETA_BUTTON_EVENT = "lutron_caseta_button_event"
BRIDGE_DEVICE_ID = "1"
MANUFACTURER = "Lutron"
ATTR_SERIAL = "serial"
ATTR_TYPE = "type"
ATTR_BUTTON_NUMBER = "button_number"
ATTR_DEVICE_NAME = "device_name"
ATTR_AREA_NAME = "area_name"
ATTR_ACTION = "action"
ACTION_PRESS = "press"
ACTION_RELEASE = "release"
CONF_TYPE = "type"
CONF_SUBTYPE = "subtype"

9
homeassistant/components/lutron_caseta/cover.py
View file

@ -12,7 +12,8 @@ from homeassistant.components.cover import (
CoverEntity,
)
from . import DOMAIN as CASETA_DOMAIN, LutronCasetaDevice
from . import LutronCasetaDevice
from .const import BRIDGE_DEVICE, BRIDGE_LEAP, DOMAIN as CASETA_DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -25,11 +26,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"""
entities = []
bridge = hass.data[CASETA_DOMAIN][config_entry.entry_id]
data = hass.data[CASETA_DOMAIN][config_entry.entry_id]
bridge = data[BRIDGE_LEAP]
bridge_device = data[BRIDGE_DEVICE]
cover_devices = bridge.get_devices_by_domain(DOMAIN)
for cover_device in cover_devices:
entity = LutronCasetaCover(cover_device, bridge)
entity = LutronCasetaCover(cover_device, bridge, bridge_device)
entities.append(entity)
async_add_entities(entities, True)

296
homeassistant/components/lutron_caseta/device_trigger.py Normal file
View file

@ -0,0 +1,296 @@
"""Provides device triggers for lutron caseta."""
import logging
from typing import List
import voluptuous as vol
from homeassistant.components.automation import AutomationActionType
from homeassistant.components.device_automation import TRIGGER_BASE_SCHEMA
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.homeassistant.triggers import event as event_trigger
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_EVENT,
CONF_PLATFORM,
CONF_TYPE,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.typing import ConfigType
from .const import (
ACTION_PRESS,
ACTION_RELEASE,
ATTR_ACTION,
ATTR_BUTTON_NUMBER,
ATTR_SERIAL,
BUTTON_DEVICES,
CONF_SUBTYPE,
DOMAIN,
LUTRON_CASETA_BUTTON_EVENT,
)
_LOGGER = logging.getLogger(__name__)
SUPPORTED_INPUTS_EVENTS_TYPES = [ACTION_PRESS, ACTION_RELEASE]
LUTRON_BUTTON_TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(SUPPORTED_INPUTS_EVENTS_TYPES),
}
)
PICO_2_BUTTON_BUTTON_TYPES = {
"on": 2,
"off": 4,
}
PICO_2_BUTTON_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_2_BUTTON_BUTTON_TYPES),
}
)
PICO_2_BUTTON_RAISE_LOWER_BUTTON_TYPES = {
"on": 2,
"off": 4,
"raise": 5,
"lower": 6,
}
PICO_2_BUTTON_RAISE_LOWER_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_2_BUTTON_RAISE_LOWER_BUTTON_TYPES),
}
)
PICO_3_BUTTON_BUTTON_TYPES = {
"on": 2,
"stop": 3,
"off": 4,
}
PICO_3_BUTTON_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_3_BUTTON_BUTTON_TYPES),
}
)
PICO_3_BUTTON_RAISE_LOWER_BUTTON_TYPES = {
"on": 2,
"stop": 3,
"off": 4,
"raise": 5,
"lower": 6,
}
PICO_3_BUTTON_RAISE_LOWER_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_3_BUTTON_RAISE_LOWER_BUTTON_TYPES),
}
)
PICO_4_BUTTON_BUTTON_TYPES = {
"button_1": 8,
"button_2": 9,
"button_3": 10,
"button_4": 11,
}
PICO_4_BUTTON_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_4_BUTTON_BUTTON_TYPES),
}
)
PICO_4_BUTTON_ZONE_BUTTON_TYPES = {
"on": 8,
"raise": 9,
"lower": 10,
"off": 11,
}
PICO_4_BUTTON_ZONE_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_4_BUTTON_ZONE_BUTTON_TYPES),
}
)
PICO_4_BUTTON_SCENE_BUTTON_TYPES = {
"button_1": 8,
"button_2": 9,
"button_3": 10,
"off": 11,
}
PICO_4_BUTTON_SCENE_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_4_BUTTON_SCENE_BUTTON_TYPES),
}
)
PICO_4_BUTTON_2_GROUP_BUTTON_TYPES = {
"group_1_button_1": 8,
"group_1_button_2": 9,
"group_2_button_1": 10,
"group_2_button_2": 11,
}
PICO_4_BUTTON_2_GROUP_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(PICO_4_BUTTON_2_GROUP_BUTTON_TYPES),
}
)
FOUR_GROUP_REMOTE_BUTTON_TYPES = {
"open_all": 2,
"stop_all": 3,
"close_all": 4,
"raise_all": 5,
"lower_all": 6,
"open_1": 10,
"stop_1": 11,
"close_1": 12,
"raise_1": 13,
"lower_1": 14,
"open_2": 18,
"stop_2": 19,
"close_2": 20,
"raise_2": 21,
"lower_2": 22,
"open_3": 26,
"stop_3": 27,
"close_3": 28,
"raise_3": 29,
"lower_3": 30,
"open_4": 34,
"stop_4": 35,
"close_4": 36,
"raise_4": 37,
"lower_4": 38,
}
FOUR_GROUP_REMOTE_TRIGGER_SCHEMA = LUTRON_BUTTON_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_SUBTYPE): vol.In(FOUR_GROUP_REMOTE_BUTTON_TYPES),
}
)
DEVICE_TYPE_SCHEMA_MAP = {
"Pico2Button": PICO_2_BUTTON_TRIGGER_SCHEMA,
"Pico2ButtonRaiseLower": PICO_2_BUTTON_RAISE_LOWER_TRIGGER_SCHEMA,
"Pico3Button": PICO_3_BUTTON_TRIGGER_SCHEMA,
"Pico3ButtonRaiseLower": PICO_3_BUTTON_RAISE_LOWER_TRIGGER_SCHEMA,
"Pico4Button": PICO_4_BUTTON_TRIGGER_SCHEMA,
"Pico4ButtonScene": PICO_4_BUTTON_SCENE_TRIGGER_SCHEMA,
"Pico4ButtonZone": PICO_4_BUTTON_ZONE_TRIGGER_SCHEMA,
"Pico4Button2Group": PICO_4_BUTTON_2_GROUP_TRIGGER_SCHEMA,
"FourGroupRemote": FOUR_GROUP_REMOTE_TRIGGER_SCHEMA,
}
DEVICE_TYPE_SUBTYPE_MAP = {
"Pico2Button": PICO_2_BUTTON_BUTTON_TYPES,
"Pico2ButtonRaiseLower": PICO_2_BUTTON_RAISE_LOWER_BUTTON_TYPES,
"Pico3Button": PICO_3_BUTTON_BUTTON_TYPES,
"Pico3ButtonRaiseLower": PICO_3_BUTTON_RAISE_LOWER_BUTTON_TYPES,
"Pico4Button": PICO_4_BUTTON_BUTTON_TYPES,
"Pico4ButtonScene": PICO_4_BUTTON_SCENE_BUTTON_TYPES,
"Pico4ButtonZone": PICO_4_BUTTON_ZONE_BUTTON_TYPES,
"Pico4Button2Group": PICO_4_BUTTON_2_GROUP_BUTTON_TYPES,
"FourGroupRemote": FOUR_GROUP_REMOTE_BUTTON_TYPES,
}
TRIGGER_SCHEMA = vol.Any(
PICO_2_BUTTON_TRIGGER_SCHEMA,
PICO_3_BUTTON_RAISE_LOWER_TRIGGER_SCHEMA,
PICO_4_BUTTON_TRIGGER_SCHEMA,
PICO_4_BUTTON_SCENE_TRIGGER_SCHEMA,
PICO_4_BUTTON_ZONE_TRIGGER_SCHEMA,
PICO_4_BUTTON_2_GROUP_TRIGGER_SCHEMA,
FOUR_GROUP_REMOTE_TRIGGER_SCHEMA,
)
async def async_validate_trigger_config(hass: HomeAssistant, config: ConfigType):
"""Validate config."""
# if device is available verify parameters against device capabilities
device = get_button_device_by_dr_id(hass, config[CONF_DEVICE_ID])
if not device:
return config
schema = DEVICE_TYPE_SCHEMA_MAP.get(device["type"])
if not schema:
raise InvalidDeviceAutomationConfig(
f"Device type {device['type']} not supported: {config[CONF_DEVICE_ID]}"
)
return schema(config)
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> List[dict]:
"""List device triggers for lutron caseta devices."""
triggers = []
device = get_button_device_by_dr_id(hass, device_id)
if not device:
raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}")
valid_buttons = DEVICE_TYPE_SUBTYPE_MAP.get(device["type"], [])
for trigger in SUPPORTED_INPUTS_EVENTS_TYPES:
for subtype in valid_buttons:
triggers.append(
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
return triggers
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: dict,
) -> CALLBACK_TYPE:
"""Attach a trigger."""
device = get_button_device_by_dr_id(hass, config[CONF_DEVICE_ID])
schema = DEVICE_TYPE_SCHEMA_MAP.get(device["type"])
valid_buttons = DEVICE_TYPE_SUBTYPE_MAP.get(device["type"])
config = schema(config)
event_config = event_trigger.TRIGGER_SCHEMA(
{
event_trigger.CONF_PLATFORM: CONF_EVENT,
event_trigger.CONF_EVENT_TYPE: LUTRON_CASETA_BUTTON_EVENT,
event_trigger.CONF_EVENT_DATA: {
ATTR_SERIAL: device["serial"],
ATTR_BUTTON_NUMBER: valid_buttons[config[CONF_SUBTYPE]],
ATTR_ACTION: config[CONF_TYPE],
},
}
)
event_config = event_trigger.TRIGGER_SCHEMA(event_config)
return await event_trigger.async_attach_trigger(
hass, event_config, action, automation_info, platform_type="device"
)
def get_button_device_by_dr_id(hass: HomeAssistant, device_id: str):
"""Get a lutron device for the given device id."""
if DOMAIN not in hass.data:
return None
for config_entry in hass.data[DOMAIN]:
button_devices = hass.data[DOMAIN][config_entry][BUTTON_DEVICES]
device = button_devices.get(device_id)
if device:
return device
return None

9
homeassistant/components/lutron_caseta/fan.py
View file

@ -13,7 +13,8 @@ from homeassistant.components.fan import (
FanEntity,
)
from . import DOMAIN as CASETA_DOMAIN, LutronCasetaDevice
from . import LutronCasetaDevice
from .const import BRIDGE_DEVICE, BRIDGE_LEAP, DOMAIN as CASETA_DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -44,11 +45,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"""
entities = []
bridge = hass.data[CASETA_DOMAIN][config_entry.entry_id]
data = hass.data[CASETA_DOMAIN][config_entry.entry_id]
bridge = data[BRIDGE_LEAP]
bridge_device = data[BRIDGE_DEVICE]
fan_devices = bridge.get_devices_by_domain(DOMAIN)
for fan_device in fan_devices:
entity = LutronCasetaFan(fan_device, bridge)
entity = LutronCasetaFan(fan_device, bridge, bridge_device)
entities.append(entity)
async_add_entities(entities, True)

9
homeassistant/components/lutron_caseta/light.py
View file

@ -11,7 +11,8 @@ from homeassistant.components.light import (
LightEntity,
)
from . import DOMAIN as CASETA_DOMAIN, LutronCasetaDevice
from . import LutronCasetaDevice
from .const import BRIDGE_DEVICE, BRIDGE_LEAP, DOMAIN as CASETA_DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -34,11 +35,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"""
entities = []
bridge = hass.data[CASETA_DOMAIN][config_entry.entry_id]
data = hass.data[CASETA_DOMAIN][config_entry.entry_id]
bridge = data[BRIDGE_LEAP]
bridge_device = data[BRIDGE_DEVICE]
light_devices = bridge.get_devices_by_domain(DOMAIN)
for light_device in light_devices:
entity = LutronCasetaLight(light_device, bridge)
entity = LutronCasetaLight(light_device, bridge, bridge_device)
entities.append(entity)
async_add_entities(entities, True)

2
homeassistant/components/lutron_caseta/manifest.json
View file

@ -3,7 +3,7 @@
"name": "Lutron Caséta",
"documentation": "https://www.home-assistant.io/integrations/lutron_caseta",
"requirements": [
"pylutron-caseta==0.8.0"
"pylutron-caseta==0.9.0", "aiolip==1.0.1"
],
"config_flow": true,
"zeroconf": ["_leap._tcp.local."],

5
homeassistant/components/lutron_caseta/scene.py
View file

@ -3,7 +3,7 @@ from typing import Any
from homeassistant.components.scene import Scene
from . import DOMAIN as CASETA_DOMAIN
from .const import BRIDGE_LEAP, DOMAIN as CASETA_DOMAIN
async def async_setup_entry(hass, config_entry, async_add_entities):
@ -14,7 +14,8 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"""
entities = []
bridge = hass.data[CASETA_DOMAIN][config_entry.entry_id]
data = hass.data[CASETA_DOMAIN][config_entry.entry_id]
bridge = data[BRIDGE_LEAP]
scenes = bridge.get_scenes()
for scene in scenes:

46
homeassistant/components/lutron_caseta/strings.json
View file

@ -26,5 +26,51 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
}
},
"device_automation": {
"trigger_subtype": {
"button_1": "First button",
"button_2": "Second button",
"button_3": "Third button",
"button_4": "Fourth button",
"group_1_button_1": "First Group first button",
"group_1_button_2": "First Group second button",
"group_2_button_1": "Second Group first button",
"group_2_button_2": "Second Group second button",
"on": "On",
"stop": "Stop (favorite)",
"off": "Off",
"raise": "Raise",
"lower": "Lower",
"open_all": "Open all",
"stop_all": "Stop all",
"close_all": "Close all",
"raise_all": "Raise all",
"lower_all": "Lower all",
"open_1": "Open 1",
"stop_1": "Stop 1",
"close_1": "Close 1",
"raise_1": "Raise 1",
"lower_1": "Lower 1",
"open_2": "Open 2",
"stop_2": "Stop 2",
"close_2": "Close 2",
"raise_2": "Raise 2",
"lower_2": "Lower 2",
"open_3": "Open 3",
"stop_3": "Stop 3",
"close_3": "Close 3",
"raise_3": "Raise 3",
"lower_3": "Lower 3",
"open_4": "Open 4",
"stop_4": "Stop 4",
"close_4": "Close 4",
"raise_4": "Raise 4",
"lower_4": "Lower 4"
},
"trigger_type": {
"press": "\"{subtype}\" pressed",
"release": "\"{subtype}\" released"
}
}
}

9
homeassistant/components/lutron_caseta/switch.py
View file

@ -3,7 +3,8 @@ import logging
from homeassistant.components.switch import DOMAIN, SwitchEntity
from . import DOMAIN as CASETA_DOMAIN, LutronCasetaDevice
from . import LutronCasetaDevice
from .const import BRIDGE_DEVICE, BRIDGE_LEAP, DOMAIN as CASETA_DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -16,11 +17,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"""
entities = []
bridge = hass.data[CASETA_DOMAIN][config_entry.entry_id]
data = hass.data[CASETA_DOMAIN][config_entry.entry_id]
bridge = data[BRIDGE_LEAP]
bridge_device = data[BRIDGE_DEVICE]
switch_devices = bridge.get_devices_by_domain(DOMAIN)
for switch_device in switch_devices:
entity = LutronCasetaLight(switch_device, bridge)
entity = LutronCasetaLight(switch_device, bridge, bridge_device)
entities.append(entity)
async_add_entities(entities, True)

46
homeassistant/components/lutron_caseta/translations/en.json
View file

@ -26,5 +26,51 @@
"title": "Automaticlly connect to the bridge"
}
}
},
"device_automation": {
"trigger_subtype": {
"button_1": "First button",
"button_2": "Second button",
"button_3": "Third button",
"button_4": "Fourth button",
"close_1": "Close 1",
"close_2": "Close 2",
"close_3": "Close 3",
"close_4": "Close 4",
"close_all": "Close all",
"group_1_button_1": "First Group first button",
"group_1_button_2": "First Group second button",
"group_2_button_1": "Second Group first button",
"group_2_button_2": "Second Group second button",
"lower": "Lower",
"lower_1": "Lower 1",
"lower_2": "Lower 2",
"lower_3": "Lower 3",
"lower_4": "Lower 4",
"lower_all": "Lower all",
"off": "Off",
"on": "On",
"open_1": "Open 1",
"open_2": "Open 2",
"open_3": "Open 3",
"open_4": "Open 4",
"open_all": "Open all",
"raise": "Raise",
"raise_1": "Raise 1",
"raise_2": "Raise 2",
"raise_3": "Raise 3",
"raise_4": "Raise 4",
"raise_all": "Raise all",
"stop": "Stop (favorite)",
"stop_1": "Stop 1",
"stop_2": "Stop 2",
"stop_3": "Stop 3",
"stop_4": "Stop 4",
"stop_all": "Stop all"
},
"trigger_type": {
"press": "\"{subtype}\" pressed",
"release": "\"{subtype}\" released"
}
}
}

5
requirements_all.txt
View file

@ -196,6 +196,9 @@ aiolifx==0.6.9
# homeassistant.components.lifx
aiolifx_effects==0.2.2
# homeassistant.components.lutron_caseta
aiolip==1.0.1
# homeassistant.components.keyboard_remote
aionotify==0.2.0
@ -1500,7 +1503,7 @@ pylitejet==0.1
pyloopenergy==0.2.1
# homeassistant.components.lutron_caseta
pylutron-caseta==0.8.0
pylutron-caseta==0.9.0
# homeassistant.components.lutron
pylutron==0.2.5

5
requirements_test_all.txt
View file

@ -115,6 +115,9 @@ aiohue==2.1.0
# homeassistant.components.apache_kafka
aiokafka==0.6.0
# homeassistant.components.lutron_caseta
aiolip==1.0.1
# homeassistant.components.notion
aionotion==1.1.0
@ -764,7 +767,7 @@ pylibrespot-java==0.1.0
pylitejet==0.1
# homeassistant.components.lutron_caseta
pylutron-caseta==0.8.0
pylutron-caseta==0.9.0
# homeassistant.components.mailgun
pymailgunner==1.4

346
tests/components/lutron_caseta/test_device_trigger.py Normal file
View file

@ -0,0 +1,346 @@
"""The tests for Lutron Caséta device triggers."""
import pytest
from homeassistant import setup
from homeassistant.components import automation
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.lutron_caseta import (
ATTR_ACTION,
ATTR_AREA_NAME,
ATTR_BUTTON_NUMBER,
ATTR_DEVICE_NAME,
ATTR_SERIAL,
ATTR_TYPE,
)
from homeassistant.components.lutron_caseta.const import (
BUTTON_DEVICES,
DOMAIN,
LUTRON_CASETA_BUTTON_EVENT,
MANUFACTURER,
)
from homeassistant.components.lutron_caseta.device_trigger import CONF_SUBTYPE
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.helpers import device_registry
from homeassistant.setup import async_setup_component
from tests.common import (
MockConfigEntry,
assert_lists_same,
async_get_device_automations,
async_mock_service,
mock_device_registry,
)
MOCK_BUTTON_DEVICES = [
{
"Name": "Back Hall Pico",
"ID": 2,
"Area": {"Name": "Back Hall"},
"Buttons": [
{"Number": 2},
{"Number": 3},
{"Number": 4},
{"Number": 5},
{"Number": 6},
],
"leap_name": "Back Hall_Back Hall Pico",
"type": "Pico3ButtonRaiseLower",
"model": "PJ2-3BRL-GXX-X01",
"serial": 43845548,
}
]
@pytest.fixture
def calls(hass):
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.fixture
def device_reg(hass):
"""Return an empty, loaded, registry."""
return mock_device_registry(hass)
async def _async_setup_lutron_with_picos(hass, device_reg):
"""Setups a lutron bridge with picos."""
await async_setup_component(hass, DOMAIN, {})
config_entry = MockConfigEntry(domain=DOMAIN, data={})
config_entry.add_to_hass(hass)
dr_button_devices = {}
for device in MOCK_BUTTON_DEVICES:
dr_device = device_reg.async_get_or_create(
name=device["leap_name"],
manufacturer=MANUFACTURER,
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, device["serial"])},
model=f"{device['model']} ({device[CONF_TYPE]})",
)
dr_button_devices[dr_device.id] = device
hass.data[DOMAIN][config_entry.entry_id] = {BUTTON_DEVICES: dr_button_devices}
return config_entry.entry_id
async def test_get_triggers(hass, device_reg):
"""Test we get the expected triggers from a lutron pico."""
config_entry_id = await _async_setup_lutron_with_picos(hass, device_reg)
dr_button_devices = hass.data[DOMAIN][config_entry_id][BUTTON_DEVICES]
device_id = list(dr_button_devices)[0]
expected_triggers = [
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "on",
CONF_TYPE: "press",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "stop",
CONF_TYPE: "press",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "off",
CONF_TYPE: "press",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "raise",
CONF_TYPE: "press",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "lower",
CONF_TYPE: "press",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "on",
CONF_TYPE: "release",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "stop",
CONF_TYPE: "release",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "off",
CONF_TYPE: "release",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "raise",
CONF_TYPE: "release",
},
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_PLATFORM: "device",
CONF_SUBTYPE: "lower",
CONF_TYPE: "release",
},
]
triggers = await async_get_device_automations(hass, "trigger", device_id)
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_for_invalid_device_id(hass, device_reg):
"""Test error raised for invalid lutron device_id."""
config_entry_id = await _async_setup_lutron_with_picos(hass, device_reg)
invalid_device = device_reg.async_get_or_create(
config_entry_id=config_entry_id,
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
with pytest.raises(InvalidDeviceAutomationConfig):
await async_get_device_automations(hass, "trigger", invalid_device.id)
async def test_if_fires_on_button_event(hass, calls, device_reg):
"""Test for press trigger firing."""
await setup.async_setup_component(hass, "persistent_notification", {})
config_entry_id = await _async_setup_lutron_with_picos(hass, device_reg)
dr_button_devices = hass.data[DOMAIN][config_entry_id][BUTTON_DEVICES]
device_id = list(dr_button_devices)[0]
device = dr_button_devices[device_id]
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
CONF_TYPE: "press",
CONF_SUBTYPE: "on",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_button_press"},
},
},
]
},
)
message = {
ATTR_SERIAL: device.get("serial"),
ATTR_TYPE: device.get("type"),
ATTR_BUTTON_NUMBER: 2,
ATTR_DEVICE_NAME: device["Name"],
ATTR_AREA_NAME: device.get("Area", {}).get("Name"),
ATTR_ACTION: "press",
}
hass.bus.async_fire(LUTRON_CASETA_BUTTON_EVENT, message)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_button_press"
async def test_validate_trigger_config_no_device(hass, calls, device_reg):
"""Test for no press with no device."""
await setup.async_setup_component(hass, "persistent_notification", {})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: "no_device",
CONF_TYPE: "press",
CONF_SUBTYPE: "on",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_button_press"},
},
},
]
},
)
message = {
ATTR_SERIAL: "123",
ATTR_TYPE: "any",
ATTR_BUTTON_NUMBER: 3,
ATTR_DEVICE_NAME: "any",
ATTR_AREA_NAME: "area",
ATTR_ACTION: "press",
}
hass.bus.async_fire(LUTRON_CASETA_BUTTON_EVENT, message)
await hass.async_block_till_done()
assert len(calls) == 0
async def test_validate_trigger_config_unknown_device(hass, calls, device_reg):
"""Test for no press with an unknown device."""
await setup.async_setup_component(hass, "persistent_notification", {})
config_entry_id = await _async_setup_lutron_with_picos(hass, device_reg)
dr_button_devices = hass.data[DOMAIN][config_entry_id][BUTTON_DEVICES]
device_id = list(dr_button_devices)[0]
device = dr_button_devices[device_id]
device["type"] = "unknown"
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
CONF_TYPE: "press",
CONF_SUBTYPE: "on",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_button_press"},
},
},
]
},
)
message = {
ATTR_SERIAL: "123",
ATTR_TYPE: "any",
ATTR_BUTTON_NUMBER: 3,
ATTR_DEVICE_NAME: "any",
ATTR_AREA_NAME: "area",
ATTR_ACTION: "press",
}
hass.bus.async_fire(LUTRON_CASETA_BUTTON_EVENT, message)
await hass.async_block_till_done()
assert len(calls) == 0
async def test_validate_trigger_invalid_triggers(hass, device_reg):
"""Test for click_event with invalid triggers."""
notification_calls = async_mock_service(hass, "persistent_notification", "create")
config_entry_id = await _async_setup_lutron_with_picos(hass, device_reg)
dr_button_devices = hass.data[DOMAIN][config_entry_id][BUTTON_DEVICES]
device_id = list(dr_button_devices)[0]
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
CONF_TYPE: "press",
CONF_SUBTYPE: "on",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_button_press"},
},
},
]
},
)
assert len(notification_calls) == 1
assert (
"The following integrations and platforms could not be set up"
in notification_calls[0].data["message"]
)