Add button platform to Matter integration (#123665)

* Add files via upload

* add test

* add discovery schemas for operational state commands

* tests

* add filter resets

* add filter reset buttons

* Apply suggestions from code review

* tweak test

---------

Co-authored-by: Marcel van der Veldt <m.vanderveldt@outlook.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
pull/126690/head
jvmahon 2024-09-24 16:07:29 -04:00 committed by GitHub
parent 9a4a66b33f
commit 5e2955845a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 288 additions and 11 deletions

View File

@ -0,0 +1,149 @@
"""Matter Button platform."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from chip.clusters import Objects as clusters
from homeassistant.components.button import (
ButtonDeviceClass,
ButtonEntity,
ButtonEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .entity import MatterEntity, MatterEntityDescription
from .helpers import get_matter
from .models import MatterDiscoverySchema
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Matter Button platform."""
matter = get_matter(hass)
matter.register_platform_handler(Platform.BUTTON, async_add_entities)
@dataclass(frozen=True)
class MatterButtonEntityDescription(ButtonEntityDescription, MatterEntityDescription):
"""Describe Matter Button entities."""
command: Callable[[], Any] | None = None
class MatterCommandButton(MatterEntity, ButtonEntity):
"""Representation of a Matter Button entity."""
entity_description: MatterButtonEntityDescription
async def async_press(self) -> None:
"""Handle the button press leveraging a Matter command."""
if TYPE_CHECKING:
assert self.entity_description.command is not None
await self.matter_client.send_device_command(
node_id=self._endpoint.node.node_id,
endpoint_id=self._endpoint.endpoint_id,
command=self.entity_description.command(),
)
# Discovery schema(s) to map Matter Attributes to HA entities
DISCOVERY_SCHEMAS = [
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="IdentifyButton",
entity_category=EntityCategory.CONFIG,
device_class=ButtonDeviceClass.IDENTIFY,
command=lambda: clusters.Identify.Commands.Identify(identifyTime=15),
),
entity_class=MatterCommandButton,
required_attributes=(clusters.Identify.Attributes.AcceptedCommandList,),
value_contains=clusters.Identify.Commands.Identify.command_id,
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="OperationalStatePauseButton",
translation_key="pause",
command=clusters.OperationalState.Commands.Pause,
),
entity_class=MatterCommandButton,
required_attributes=(clusters.OperationalState.Attributes.AcceptedCommandList,),
value_contains=clusters.OperationalState.Commands.Pause.command_id,
allow_multi=True,
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="OperationalStateResumeButton",
translation_key="resume",
command=clusters.OperationalState.Commands.Resume,
),
entity_class=MatterCommandButton,
required_attributes=(clusters.OperationalState.Attributes.AcceptedCommandList,),
value_contains=clusters.OperationalState.Commands.Resume.command_id,
allow_multi=True,
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="OperationalStateStartButton",
translation_key="start",
command=clusters.OperationalState.Commands.Start,
),
entity_class=MatterCommandButton,
required_attributes=(clusters.OperationalState.Attributes.AcceptedCommandList,),
value_contains=clusters.OperationalState.Commands.Start.command_id,
allow_multi=True,
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="OperationalStateStopButton",
translation_key="stop",
command=clusters.OperationalState.Commands.Stop,
),
entity_class=MatterCommandButton,
required_attributes=(clusters.OperationalState.Attributes.AcceptedCommandList,),
value_contains=clusters.OperationalState.Commands.Stop.command_id,
allow_multi=True,
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="HepaFilterMonitoringResetButton",
translation_key="reset_filter_condition",
command=clusters.HepaFilterMonitoring.Commands.ResetCondition,
),
entity_class=MatterCommandButton,
required_attributes=(
clusters.HepaFilterMonitoring.Attributes.AcceptedCommandList,
),
value_contains=clusters.HepaFilterMonitoring.Commands.ResetCondition.command_id,
allow_multi=True,
),
MatterDiscoverySchema(
platform=Platform.BUTTON,
entity_description=MatterButtonEntityDescription(
key="ActivatedCarbonFilterMonitoringResetButton",
translation_key="reset_filter_condition",
command=clusters.ActivatedCarbonFilterMonitoring.Commands.ResetCondition,
),
entity_class=MatterCommandButton,
required_attributes=(
clusters.ActivatedCarbonFilterMonitoring.Attributes.AcceptedCommandList,
),
value_contains=clusters.ActivatedCarbonFilterMonitoring.Commands.ResetCondition.command_id,
allow_multi=True,
),
]

View File

@ -11,6 +11,7 @@ from homeassistant.const import Platform
from homeassistant.core import callback
from .binary_sensor import DISCOVERY_SCHEMAS as BINARY_SENSOR_SCHEMAS
from .button import DISCOVERY_SCHEMAS as BUTTON_SCHEMAS
from .climate import DISCOVERY_SCHEMAS as CLIMATE_SENSOR_SCHEMAS
from .cover import DISCOVERY_SCHEMAS as COVER_SCHEMAS
from .event import DISCOVERY_SCHEMAS as EVENT_SCHEMAS
@ -26,6 +27,7 @@ from .update import DISCOVERY_SCHEMAS as UPDATE_SCHEMAS
DISCOVERY_SCHEMAS: dict[Platform, list[MatterDiscoverySchema]] = {
Platform.BINARY_SENSOR: BINARY_SENSOR_SCHEMAS,
Platform.BUTTON: BUTTON_SCHEMAS,
Platform.CLIMATE: CLIMATE_SENSOR_SCHEMAS,
Platform.COVER: COVER_SCHEMAS,
Platform.EVENT: EVENT_SCHEMAS,
@ -114,6 +116,16 @@ def async_discover_entities(
):
continue
# check for required value in (primary) attribute
if schema.value_contains is not None and (
(primary_attribute := next((x for x in schema.required_attributes), None))
is None
or (value := endpoint.get_attribute_value(None, primary_attribute)) is None
or not isinstance(value, list)
or schema.value_contains not in value
):
continue
# all checks passed, this value belongs to an entity
attributes_to_watch = list(schema.required_attributes)

View File

@ -2,7 +2,6 @@
from __future__ import annotations
from abc import abstractmethod
from collections.abc import Callable
from dataclasses import dataclass
from functools import cached_property
@ -158,7 +157,6 @@ class MatterEntity(Entity):
self.async_write_ha_state()
@callback
@abstractmethod
def _update_from_device(self) -> None:
"""Update data from Matter device."""

View File

@ -5,6 +5,20 @@
"default": "mdi:bell-off"
}
},
"button": {
"pause": {
"default": "mdi:pause"
},
"resume": {
"default": "mdi:play-pause"
},
"start": {
"default": "mdi:play"
},
"stop": {
"default": "mdi:stop"
}
},
"fan": {
"fan": {
"state_attributes": {

View File

@ -3,7 +3,7 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import TypedDict
from typing import Any, TypedDict
from chip.clusters import Objects as clusters
from chip.clusters.Objects import Cluster, ClusterAttributeDescriptor
@ -108,6 +108,11 @@ class MatterDiscoverySchema:
# are not discovered by other entities
optional_attributes: tuple[type[ClusterAttributeDescriptor], ...] | None = None
# [optional] the primary attribute value must contain this value
# for example for the AcceptedCommandList
# NOTE: only works for list values
value_contains: Any | None = None
# [optional] bool to specify if this primary value may be discovered
# by multiple platforms
allow_multi: bool = False

View File

@ -77,6 +77,23 @@
"name": "Muted"
}
},
"button": {
"pause": {
"name": "[%key:common::action::pause%]"
},
"resume": {
"name": "Resume"
},
"start": {
"name": "[%key:common::action::start%]"
},
"stop": {
"name": "[%key:common::action::stop%]"
},
"reset_filter_condition": {
"name": "Reset filter condition"
}
},
"climate": {
"thermostat": {
"name": "Thermostat"

View File

@ -305,13 +305,6 @@
"0/65/65528": [],
"0/65/65529": [],
"0/65/65531": [0, 65528, 65529, 65531, 65532, 65533],
"1/3/0": 0,
"1/3/1": 0,
"1/3/65532": 0,
"1/3/65533": 4,
"1/3/65528": [],
"1/3/65529": [0, 64],
"1/3/65531": [0, 1, 65528, 65529, 65531, 65532, 65533],
"1/4/0": 128,
"1/4/65532": 1,
"1/4/65533": 4,

View File

@ -444,7 +444,7 @@
"1/96/65532": 0,
"1/96/65533": 1,
"1/96/65528": [4],
"1/96/65529": [0, 1, 2, 3],
"1/96/65529": [0, 1, 2],
"1/96/65531": [0, 1, 3, 4, 5, 65528, 65529, 65531, 65532, 65533],
"2/29/0": [
{

View File

@ -0,0 +1,89 @@
"""Test Matter switches."""
from unittest.mock import MagicMock, call
from chip.clusters import Objects as clusters
from matter_server.client.models.node import MatterNode
import pytest
from homeassistant.core import HomeAssistant
from .common import setup_integration_with_node_fixture
@pytest.fixture(name="powerplug_node")
async def powerplug_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Powerplug node."""
return await setup_integration_with_node_fixture(
hass, "eve-energy-plug", matter_client
)
@pytest.fixture(name="dishwasher_node")
async def dishwasher_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for an dishwasher node."""
return await setup_integration_with_node_fixture(
hass, "silabs-dishwasher", matter_client
)
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
async def test_identify_button(
hass: HomeAssistant,
matter_client: MagicMock,
powerplug_node: MatterNode,
) -> None:
"""Test button entity is created for a Matter Identify Cluster."""
state = hass.states.get("button.eve_energy_plug_identify")
assert state
assert state.attributes["friendly_name"] == "Eve Energy Plug Identify"
# test press action
await hass.services.async_call(
"button",
"press",
{
"entity_id": "button.eve_energy_plug_identify",
},
blocking=True,
)
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=powerplug_node.node_id,
endpoint_id=1,
command=clusters.Identify.Commands.Identify(identifyTime=15),
)
async def test_operational_state_buttons(
hass: HomeAssistant,
matter_client: MagicMock,
dishwasher_node: MatterNode,
) -> None:
"""Test if button entities are created for operational state commands."""
assert hass.states.get("button.dishwasher_pause")
assert hass.states.get("button.dishwasher_start")
assert hass.states.get("button.dishwasher_stop")
# resume may not be disocvered as its missing in the supported command list
assert hass.states.get("button.dishwasher_resume") is None
# test press action
await hass.services.async_call(
"button",
"press",
{
"entity_id": "button.dishwasher_pause",
},
blocking=True,
)
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=dishwasher_node.node_id,
endpoint_id=1,
command=clusters.OperationalState.Commands.Pause(),
)