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

Add event platform for Shelly gen1 devices (#100655)

Browse Source 
* Initial commit

* Use description.key

* Add translations

* Check event_types

* Rename input_id to channel

* Fix removeal confition

* Add tests

* Sort classes and consts

* Use ShellyBlockEntity class

* Update tests

* Update homeassistant/components/shelly/event.py

Co-authored-by: Shay Levy <levyshay1@gmail.com>

---------

Co-authored-by: Shay Levy <levyshay1@gmail.com>
This commit is contained in:
Maciej Bieniek 2023-09-23 14:03:57 +00:00 committed by GitHub
parent 173b70c850
commit 5c5dff034c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 166 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -58,6 +58,7 @@ BLOCK_PLATFORMS: Final = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
Platform.COVER,
Platform.EVENT,
Platform.LIGHT,
Platform.SENSOR,
Platform.SWITCH,

18
homeassistant/components/shelly/coordinator.py
View File

@ -170,6 +170,7 @@ class ShellyBlockCoordinator(ShellyCoordinatorBase[BlockDevice]):
self._last_input_events_count: dict = {}
self._last_target_temp: float | None = None
self._push_update_failures: int = 0
self._input_event_listeners: list[Callable[[dict[str, Any]], None]] = []
entry.async_on_unload(
self.async_add_listener(self._async_device_updates_handler)
@ -178,6 +179,19 @@ class ShellyBlockCoordinator(ShellyCoordinatorBase[BlockDevice]):
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
)
@callback
def async_subscribe_input_events(
self, input_event_callback: Callable[[dict[str, Any]], None]
) -> CALLBACK_TYPE:
"""Subscribe to input events."""
def _unsubscribe() -> None:
self._input_event_listeners.remove(input_event_callback)
self._input_event_listeners.append(input_event_callback)
return _unsubscribe
@callback
def _async_device_updates_handler(self) -> None:
"""Handle device updates."""
@ -242,6 +256,10 @@ class ShellyBlockCoordinator(ShellyCoordinatorBase[BlockDevice]):
continue
if event_type in INPUTS_EVENTS_DICT:
for event_callback in self._input_event_listeners:
event_callback(
{"channel": channel, "event": INPUTS_EVENTS_DICT[event_type]}
)
self.hass.bus.async_fire(
EVENT_SHELLY_CLICK,
{

106
homeassistant/components/shelly/event.py
View File

@ -3,7 +3,9 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, Final
from typing import TYPE_CHECKING, Any, Final
from aioshelly.block_device import Block
from homeassistant.components.event import (
DOMAIN as EVENT_DOMAIN,
@ -17,25 +19,46 @@ from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, Device
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import RPC_INPUTS_EVENTS_TYPES
from .coordinator import ShellyRpcCoordinator, get_entry_data
from .const import (
BASIC_INPUTS_EVENTS_TYPES,
RPC_INPUTS_EVENTS_TYPES,
SHIX3_1_INPUTS_EVENTS_TYPES,
)
from .coordinator import ShellyBlockCoordinator, ShellyRpcCoordinator, get_entry_data
from .entity import ShellyBlockEntity
from .utils import (
async_remove_shelly_entity,
get_device_entry_gen,
get_rpc_entity_name,
get_rpc_key_instances,
is_block_momentary_input,
is_rpc_momentary_input,
)
@dataclass
class ShellyEventDescription(EventEntityDescription):
class ShellyBlockEventDescription(EventEntityDescription):
"""Class to describe Shelly event."""
removal_condition: Callable[[dict, Block], bool] | None = None
@dataclass
class ShellyRpcEventDescription(EventEntityDescription):
"""Class to describe Shelly event."""
removal_condition: Callable[[dict, dict, str], bool] | None = None
RPC_EVENT: Final = ShellyEventDescription(
BLOCK_EVENT: Final = ShellyBlockEventDescription(
key="input",
translation_key="input",
device_class=EventDeviceClass.BUTTON,
removal_condition=lambda settings, block: not is_block_momentary_input(
settings, block, True
),
)
RPC_EVENT: Final = ShellyRpcEventDescription(
key="input",
translation_key="input",
device_class=EventDeviceClass.BUTTON,
@ -52,11 +75,15 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up sensors for device."""
entities: list[ShellyBlockEvent | ShellyRpcEvent] = []
coordinator: ShellyRpcCoordinator | ShellyBlockCoordinator | None = None
if get_device_entry_gen(config_entry) == 2:
coordinator = get_entry_data(hass)[config_entry.entry_id].rpc
assert coordinator
if TYPE_CHECKING:
assert coordinator
entities = []
key_instances = get_rpc_key_instances(coordinator.device.status, RPC_EVENT.key)
for key in key_instances:
@ -67,21 +94,80 @@ async def async_setup_entry(
async_remove_shelly_entity(hass, EVENT_DOMAIN, unique_id)
else:
entities.append(ShellyRpcEvent(coordinator, key, RPC_EVENT))
else:
coordinator = get_entry_data(hass)[config_entry.entry_id].block
if TYPE_CHECKING:
assert coordinator
assert coordinator.device.blocks
async_add_entities(entities)
for block in coordinator.device.blocks:
if (
"inputEvent" not in block.sensor_ids
or "inputEventCnt" not in block.sensor_ids
):
continue
if BLOCK_EVENT.removal_condition and BLOCK_EVENT.removal_condition(
coordinator.device.settings, block
):
channel = int(block.channel or 0) + 1
unique_id = f"{coordinator.mac}-{block.description}-{channel}"
async_remove_shelly_entity(hass, EVENT_DOMAIN, unique_id)
else:
entities.append(ShellyBlockEvent(coordinator, block, BLOCK_EVENT))
async_add_entities(entities)
class ShellyBlockEvent(ShellyBlockEntity, EventEntity):
"""Represent Block event entity."""
_attr_should_poll = False
entity_description: ShellyBlockEventDescription
def __init__(
self,
coordinator: ShellyBlockCoordinator,
block: Block,
description: ShellyBlockEventDescription,
) -> None:
"""Initialize Shelly entity."""
super().__init__(coordinator, block)
self.channel = channel = int(block.channel or 0) + 1
self._attr_unique_id = f"{super().unique_id}-{channel}"
if coordinator.model == "SHIX3-1":
self._attr_event_types = list(SHIX3_1_INPUTS_EVENTS_TYPES)
else:
self._attr_event_types = list(BASIC_INPUTS_EVENTS_TYPES)
self.entity_description = description
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
self.coordinator.async_subscribe_input_events(self._async_handle_event)
)
@callback
def _async_handle_event(self, event: dict[str, Any]) -> None:
"""Handle the demo button event."""
if event["channel"] == self.channel:
self._trigger_event(event["event"])
self.async_write_ha_state()
class ShellyRpcEvent(CoordinatorEntity[ShellyRpcCoordinator], EventEntity):
"""Represent RPC event entity."""
_attr_should_poll = False
entity_description: ShellyEventDescription
entity_description: ShellyRpcEventDescription
def __init__(
self,
coordinator: ShellyRpcCoordinator,
key: str,
description: ShellyEventDescription,
description: ShellyRpcEventDescription,
) -> None:
"""Initialize Shelly entity."""
super().__init__(coordinator)

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

@ -106,9 +106,15 @@
"btn_down": "Button down",
"btn_up": "Button up",
"double_push": "Double push",
"double": "Double push",
"long_push": "Long push",
"long_single": "Long push and then short push",
"long": "Long push",
"single_long": "Short push and then long push",
"single_push": "Single push",
"triple_push": "Triple push"
"single": "Single push",
"triple_push": "Triple push",
"triple": "Triple push"
}
}
}

44
tests/components/shelly/test_event.py
View File

@ -15,6 +15,8 @@ from homeassistant.helpers.entity_registry import async_get
from . import init_integration, inject_rpc_device_event, register_entity
DEVICE_BLOCK_ID = 4
async def test_rpc_button(hass: HomeAssistant, mock_rpc_device, monkeypatch) -> None:
"""Test RPC device event."""
@ -68,3 +70,45 @@ async def test_rpc_event_removal(
await init_integration(hass, 2)
assert registry.async_get(entity_id) is None
async def test_block_event(hass: HomeAssistant, monkeypatch, mock_block_device) -> None:
"""Test block device event."""
await init_integration(hass, 1)
entity_id = "event.test_name_channel_1"
registry = async_get(hass)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_UNKNOWN
assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(["single", "long"])
assert state.attributes.get(ATTR_EVENT_TYPE) is None
assert state.attributes.get(ATTR_DEVICE_CLASS) == EventDeviceClass.BUTTON
entry = registry.async_get(entity_id)
assert entry
assert entry.unique_id == "123456789ABC-relay_0-1"
monkeypatch.setattr(
mock_block_device.blocks[DEVICE_BLOCK_ID],
"sensor_ids",
{"inputEvent": "L", "inputEventCnt": 0},
)
monkeypatch.setattr(mock_block_device.blocks[DEVICE_BLOCK_ID], "inputEvent", "L")
mock_block_device.mock_update()
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes.get(ATTR_EVENT_TYPE) == "long"
async def test_block_event_shix3_1(hass: HomeAssistant, mock_block_device) -> None:
"""Test block device event for SHIX3-1."""
await init_integration(hass, 1, model="SHIX3-1")
entity_id = "event.test_name_channel_1"
state = hass.states.get(entity_id)
assert state
assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(
["double", "long", "long_single", "single", "single_long", "triple"]
)