jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add logbook and device trigger platforms to Shelly (#44020)

Browse Source 
* Add logbook and device trigger platforms to Shelly

Add `logbook` platform for describing “shelly.click” event
Add `device_trigger` platform for adding automation based on click events:

Example of logbook event:
Shelly 'single' click event for Test I3 channel 3 was fired.
(Test I3 is the name of the device)

Example of automation triggers:
First button triple clicked
First button long clicked and then single clicked
First button double clicked
First button long clicked
First button single clicked and then long clicked
First button single clicked
Second button triple clicked
..
Second button single clicked

* Fix codespell

* Remove pylint added for debug

* Add tests

* Rebase

* Fix Rebase & Apply PR review suggestions

Fix tests after rebasing
Use `INPUTS_EVENTS_DICT` for input triggers
Apply PR suggestions
This commit is contained in:
Shay Levy 2021-01-05 00:10:42 +02:00 committed by GitHub
parent 773d95251e
commit 76537305e2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 639 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
homeassistant/components/shelly/__init__.py
View File

@ -9,6 +9,7 @@ import async_timeout
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_DEVICE_ID,
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
@ -25,10 +26,14 @@ from homeassistant.helpers import (
from .const import (
AIOSHELLY_DEVICE_TIMEOUT_SEC,
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION,
COAP,
DATA_CONFIG_ENTRY,
DOMAIN,
EVENT_SHELLY_CLICK,
INPUTS_EVENTS_DICT,
POLLING_TIMEOUT_MULTIPLIER,
REST,
@ -170,12 +175,12 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
if event_type in INPUTS_EVENTS_DICT:
self.hass.bus.async_fire(
"shelly.click",
EVENT_SHELLY_CLICK,
{
"device_id": self.device_id,
"device": self.device.settings["device"]["hostname"],
"channel": channel,
"click_type": INPUTS_EVENTS_DICT[event_type],
ATTR_DEVICE_ID: self.device_id,
ATTR_DEVICE: self.device.settings["device"]["hostname"],
ATTR_CHANNEL: channel,
ATTR_CLICK_TYPE: INPUTS_EVENTS_DICT[event_type],
},
)
else:

35
homeassistant/components/shelly/const.py
View File

@ -35,3 +35,38 @@ INPUTS_EVENTS_DICT = {
# List of battery devices that maintain a permanent WiFi connection
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION = ["SHMOS-01"]
EVENT_SHELLY_CLICK = "shelly.click"
ATTR_CLICK_TYPE = "click_type"
ATTR_CHANNEL = "channel"
ATTR_DEVICE = "device"
CONF_SUBTYPE = "subtype"
BASIC_INPUTS_EVENTS_TYPES = {
"single",
"long",
}
SHBTN_1_INPUTS_EVENTS_TYPES = {
"single",
"double",
"triple",
"long",
}
SUPPORTED_INPUTS_EVENTS_TYPES = SHIX3_1_INPUTS_EVENTS_TYPES = {
"single",
"double",
"triple",
"long",
"single_long",
"long_single",
}
INPUTS_EVENTS_SUBTYPES = {
"button": 1,
"button1": 1,
"button2": 2,
"button3": 3,
}

110
homeassistant/components/shelly/device_trigger.py Normal file
View File

@ -0,0 +1,110 @@
"""Provides device triggers for Shelly."""
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 (
ATTR_DEVICE_ID,
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 (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
CONF_SUBTYPE,
DOMAIN,
EVENT_SHELLY_CLICK,
INPUTS_EVENTS_SUBTYPES,
SUPPORTED_INPUTS_EVENTS_TYPES,
)
from .utils import get_device_wrapper, get_input_triggers
TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(SUPPORTED_INPUTS_EVENTS_TYPES),
vol.Required(CONF_SUBTYPE): vol.In(INPUTS_EVENTS_SUBTYPES),
}
)
async def async_validate_trigger_config(hass, config):
"""Validate config."""
config = TRIGGER_SCHEMA(config)
# if device is available verify parameters against device capabilities
wrapper = get_device_wrapper(hass, config[CONF_DEVICE_ID])
if not wrapper:
return config
trigger = (config[CONF_TYPE], config[CONF_SUBTYPE])
for block in wrapper.device.blocks:
input_triggers = get_input_triggers(wrapper.device, block)
if trigger in input_triggers:
return config
raise InvalidDeviceAutomationConfig(
f"Invalid ({CONF_TYPE},{CONF_SUBTYPE}): {trigger}"
)
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> List[dict]:
"""List device triggers for Shelly devices."""
triggers = []
wrapper = get_device_wrapper(hass, device_id)
if not wrapper:
raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}")
for block in wrapper.device.blocks:
input_triggers = get_input_triggers(wrapper.device, block)
for trigger, subtype in input_triggers:
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."""
config = TRIGGER_SCHEMA(config)
event_config = event_trigger.TRIGGER_SCHEMA(
{
event_trigger.CONF_PLATFORM: CONF_EVENT,
event_trigger.CONF_EVENT_TYPE: EVENT_SHELLY_CLICK,
event_trigger.CONF_EVENT_DATA: {
ATTR_DEVICE_ID: config[CONF_DEVICE_ID],
ATTR_CHANNEL: INPUTS_EVENTS_SUBTYPES[config[CONF_SUBTYPE]],
ATTR_CLICK_TYPE: 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"
)

37
homeassistant/components/shelly/logbook.py Normal file
View File

@ -0,0 +1,37 @@
"""Describe Shelly logbook events."""
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import callback
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
DOMAIN,
EVENT_SHELLY_CLICK,
)
from .utils import get_device_name, get_device_wrapper
@callback
def async_describe_events(hass, async_describe_event):
"""Describe logbook events."""
@callback
def async_describe_shelly_click_event(event):
"""Describe shelly.click logbook event."""
wrapper = get_device_wrapper(hass, event.data[ATTR_DEVICE_ID])
if wrapper:
device_name = get_device_name(wrapper.device)
else:
device_name = event.data[ATTR_DEVICE]
channel = event.data[ATTR_CHANNEL]
click_type = event.data[ATTR_CLICK_TYPE]
return {
"name": "Shelly",
"message": f"'{click_type}' click event for {device_name} channel {channel} was fired.",
}
async_describe_event(DOMAIN, EVENT_SHELLY_CLICK, async_describe_shelly_click_event)

16
homeassistant/components/shelly/strings.json
View File

@ -27,5 +27,21 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"unsupported_firmware": "The device is using an unsupported firmware version."
}
},
"device_automation":{
"trigger_subtype": {
"button": "Button",
"button1": "First button",
"button2": "Second button",
"button3": "Third button"
},
"trigger_type": {
"single": "{subtype} single clicked",
"double": "{subtype} double clicked",
"triple": "{subtype} triple clicked",
"long": " {subtype} long clicked",
"single_long": "{subtype} single clicked and then long clicked",
"long_single": "{subtype} long clicked and then single clicked"
}
}
}

18
homeassistant/components/shelly/translations/en.json
View File

@ -27,5 +27,21 @@
"description": "Before set up, battery-powered devices must be woken up by pressing the button on the device."
}
}
},
"device_automation": {
"trigger_subtype": {
"button": "Button",
"button1": "First button",
"button2": "Second button",
"button3": "Third button"
},
"trigger_type": {
"single": "{subtype} single clicked",
"double": "{subtype} double clicked",
"triple": "{subtype} triple clicked",
"long":" {subtype} long clicked",
"single_long": "{subtype} single clicked and then long clicked",
"long_single": "{subtype} long clicked and then single clicked"
}
}
}
}

145
homeassistant/components/shelly/utils.py
View File

@ -2,14 +2,22 @@
from datetime import timedelta
import logging
from typing import Optional
from typing import List, Optional, Tuple
import aioshelly
from homeassistant.const import TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.core import HomeAssistant
from homeassistant.util.dt import parse_datetime, utcnow
from .const import DOMAIN
from .const import (
BASIC_INPUTS_EVENTS_TYPES,
COAP,
DATA_CONFIG_ENTRY,
DOMAIN,
SHBTN_1_INPUTS_EVENTS_TYPES,
SHIX3_1_INPUTS_EVENTS_TYPES,
)
_LOGGER = logging.getLogger(__name__)
@ -35,54 +43,76 @@ def get_device_name(device: aioshelly.Device) -> str:
return device.settings["name"] or device.settings["device"]["hostname"]
def get_number_of_channels(device: aioshelly.Device, block: aioshelly.Block) -> int:
"""Get number of channels for block type."""
channels = None
if block.type == "input":
# Shelly Dimmer/1L has two input channels and missing "num_inputs"
if device.settings["device"]["type"] in ["SHDM-1", "SHDM-2", "SHSW-L"]:
channels = 2
else:
channels = device.shelly.get("num_inputs")
elif block.type == "emeter":
channels = device.shelly.get("num_emeters")
elif block.type in ["relay", "light"]:
channels = device.shelly.get("num_outputs")
elif block.type in ["roller", "device"]:
channels = 1
return channels or 1
def get_entity_name(
device: aioshelly.Device,
block: aioshelly.Block,
description: Optional[str] = None,
) -> str:
"""Naming for switch and sensors."""
entity_name = get_device_name(device)
if block:
channels = None
if block.type == "input":
# Shelly Dimmer/1L has two input channels and missing "num_inputs"
if device.settings["device"]["type"] in ["SHDM-1", "SHDM-2", "SHSW-L"]:
channels = 2
else:
channels = device.shelly.get("num_inputs")
elif block.type == "emeter":
channels = device.shelly.get("num_emeters")
elif block.type in ["relay", "light"]:
channels = device.shelly.get("num_outputs")
elif block.type in ["roller", "device"]:
channels = 1
channels = channels or 1
if channels > 1 and block.type != "device":
entity_name = None
mode = block.type + "s"
if mode in device.settings:
entity_name = device.settings[mode][int(block.channel)].get("name")
if not entity_name:
if device.settings["device"]["type"] == "SHEM-3":
base = ord("A")
else:
base = ord("1")
entity_name = (
f"{get_device_name(device)} channel {chr(int(block.channel)+base)}"
)
channel_name = get_device_channel_name(device, block)
if description:
entity_name = f"{entity_name} {description}"
return f"{channel_name} {description}"
return entity_name
return channel_name
def get_device_channel_name(
device: aioshelly.Device,
block: aioshelly.Block,
) -> str:
"""Get name based on device and channel name."""
entity_name = get_device_name(device)
if (
not block
or block.type == "device"
or get_number_of_channels(device, block) == 1
):
return entity_name
channel_name = None
mode = block.type + "s"
if mode in device.settings:
channel_name = device.settings[mode][int(block.channel)].get("name")
if channel_name:
return channel_name
if device.settings["device"]["type"] == "SHEM-3":
base = ord("A")
else:
base = ord("1")
return f"{entity_name} channel {chr(int(block.channel)+base)}"
def is_momentary_input(settings: dict, block: aioshelly.Block) -> bool:
"""Return true if input button settings is set to a momentary type."""
# Shelly Button type is fixed to momentary and no btn_type
if settings["device"]["type"] == "SHBTN-1":
return True
button = settings.get("relays") or settings.get("lights") or settings.get("inputs")
# Shelly 1L has two button settings in the first channel
@ -108,3 +138,44 @@ def get_device_uptime(status: dict, last_uptime: str) -> str:
return uptime.replace(microsecond=0).isoformat()
return last_uptime
def get_input_triggers(
device: aioshelly.Device, block: aioshelly.Block
) -> List[Tuple[str, str]]:
"""Return list of input triggers for block."""
if "inputEvent" not in block.sensor_ids or "inputEventCnt" not in block.sensor_ids:
return []
if not is_momentary_input(device.settings, block):
return []
triggers = []
if block.type == "device" or get_number_of_channels(device, block) == 1:
subtype = "button"
else:
subtype = f"button{int(block.channel)+1}"
if device.settings["device"]["type"] == "SHBTN-1":
trigger_types = SHBTN_1_INPUTS_EVENTS_TYPES
elif device.settings["device"]["type"] == "SHIX3-1":
trigger_types = SHIX3_1_INPUTS_EVENTS_TYPES
else:
trigger_types = BASIC_INPUTS_EVENTS_TYPES
for trigger_type in trigger_types:
triggers.append((trigger_type, subtype))
return triggers
def get_device_wrapper(hass: HomeAssistant, device_id: str):
"""Get a Shelly device wrapper for the given device id."""
for config_entry in hass.data[DOMAIN][DATA_CONFIG_ENTRY]:
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry][COAP]
if wrapper.device_id == device_id:
return wrapper
return None

72
tests/components/shelly/conftest.py
View File

@ -1,11 +1,81 @@
"""Test configuration for Shelly."""
from unittest.mock import patch
from unittest.mock import Mock, patch
import pytest
from homeassistant.components.shelly import ShellyDeviceWrapper
from homeassistant.components.shelly.const import (
COAP,
DATA_CONFIG_ENTRY,
DOMAIN,
EVENT_SHELLY_CLICK,
)
from homeassistant.core import callback as ha_callback
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, async_mock_service, mock_device_registry
MOCK_SETTINGS = {
"name": "Test name",
"device": {
"mac": "test-mac",
"hostname": "test-host",
"type": "SHSW-25",
"num_outputs": 2,
},
"coiot": {"update_period": 15},
"fw": "20201124-092159/v1.9.0@57ac4ad8",
"relays": [{"btn_type": "momentary"}, {"btn_type": "toggle"}],
}
MOCK_BLOCKS = [
Mock(sensor_ids={"inputEvent": "S", "inputEventCnt": 2}, channel="0", type="relay")
]
@pytest.fixture(autouse=True)
def mock_coap():
"""Mock out coap."""
with patch("homeassistant.components.shelly.get_coap_context"):
yield
@pytest.fixture
def device_reg(hass):
"""Return an empty, loaded, registry."""
return mock_device_registry(hass)
@pytest.fixture
def calls(hass):
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.fixture
def events(hass):
"""Yield caught shelly_click events."""
ha_events = []
hass.bus.async_listen(EVENT_SHELLY_CLICK, ha_callback(ha_events.append))
yield ha_events
@pytest.fixture
async def coap_wrapper(hass):
"""Setups a coap wrapper with mocked device."""
await async_setup_component(hass, "shelly", {})
config_entry = MockConfigEntry(domain=DOMAIN, data={})
config_entry.add_to_hass(hass)
device = Mock(blocks=MOCK_BLOCKS, settings=MOCK_SETTINGS)
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
COAP
] = ShellyDeviceWrapper(hass, config_entry, device)
await wrapper.async_setup()
return wrapper

173
tests/components/shelly/test_device_trigger.py Normal file
View File

@ -0,0 +1,173 @@
"""The tests for Shelly device triggers."""
import pytest
from homeassistant import setup
from homeassistant.components import automation
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.shelly.const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
CONF_SUBTYPE,
DOMAIN,
EVENT_SHELLY_CLICK,
)
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,
)
async def test_get_triggers(hass, coap_wrapper):
"""Test we get the expected triggers from a shelly."""
assert coap_wrapper
expected_triggers = [
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "single",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "long",
CONF_SUBTYPE: "button1",
},
]
triggers = await async_get_device_automations(
hass, "trigger", coap_wrapper.device_id
)
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_for_invalid_device_id(hass, device_reg, coap_wrapper):
"""Test error raised for invalid shelly device_id."""
assert coap_wrapper
config_entry = MockConfigEntry(domain=DOMAIN, data={})
config_entry.add_to_hass(hass)
invalid_device = device_reg.async_get_or_create(
config_entry_id=config_entry.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_click_event(hass, calls, coap_wrapper):
"""Test for click_event trigger firing."""
assert coap_wrapper
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: coap_wrapper.device_id,
CONF_TYPE: "single",
CONF_SUBTYPE: "button1",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_single_click"},
},
},
]
},
)
message = {
CONF_DEVICE_ID: coap_wrapper.device_id,
ATTR_CLICK_TYPE: "single",
ATTR_CHANNEL: 1,
}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
"""Test for click_event with no device."""
assert coap_wrapper
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: "single",
CONF_SUBTYPE: "button1",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_single_click"},
},
},
]
},
)
message = {CONF_DEVICE_ID: "no_device", ATTR_CLICK_TYPE: "single", ATTR_CHANNEL: 1}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_validate_trigger_invalid_triggers(hass, coap_wrapper):
"""Test for click_event with invalid triggers."""
assert coap_wrapper
notification_calls = async_mock_service(hass, "persistent_notification", "create")
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: coap_wrapper.device_id,
CONF_TYPE: "single",
CONF_SUBTYPE: "button3",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_single_click"},
},
},
]
},
)
assert len(notification_calls) == 1
assert (
"The following integrations and platforms could not be set up"
in notification_calls[0].data["message"]
)

62
tests/components/shelly/test_logbook.py Normal file
View File

@ -0,0 +1,62 @@
"""The tests for Shelly logbook."""
from homeassistant.components import logbook
from homeassistant.components.shelly.const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
DOMAIN,
EVENT_SHELLY_CLICK,
)
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.setup import async_setup_component
from tests.components.logbook.test_init import MockLazyEventPartialState
async def test_humanify_shelly_click_event(hass, coap_wrapper):
"""Test humanifying Shelly click event."""
assert coap_wrapper
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
entity_attr_cache = logbook.EntityAttributeCache(hass)
event1, event2 = list(
logbook.humanify(
hass,
[
MockLazyEventPartialState(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: coap_wrapper.device_id,
ATTR_DEVICE: "shellyix3-12345678",
ATTR_CLICK_TYPE: "single",
ATTR_CHANNEL: 1,
},
),
MockLazyEventPartialState(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: "no_device_id",
ATTR_DEVICE: "shellyswitch25-12345678",
ATTR_CLICK_TYPE: "long",
ATTR_CHANNEL: 2,
},
),
],
entity_attr_cache,
{},
)
)
assert event1["name"] == "Shelly"
assert event1["domain"] == DOMAIN
assert (
event1["message"] == "'single' click event for Test name channel 1 was fired."
)
assert event2["name"] == "Shelly"
assert event2["domain"] == DOMAIN
assert (
event2["message"]
== "'long' click event for shellyswitch25-12345678 channel 2 was fired."
)