Add trigger media_player.muted

homeassistant/components/media_player/icons.json

@@ -106,6 +106,9 @@
     }
   },
   "triggers": {
+    "muted": {
+      "trigger": "mdi:volume-mute"
+    },
     "stopped_playing": {
       "trigger": "mdi:stop"
     }

homeassistant/components/media_player/strings.json

@@ -380,6 +380,17 @@
   },
   "title": "Media player",
   "triggers": {
+    "muted": {
+      "description": "Triggers when a media player is muted.",
+      "description_configured": "[%key:component::media_player::triggers::muted::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::media_player::common::trigger_behavior_description%]",
+          "name": "[%key:component::media_player::common::trigger_behavior_name%]"
+        }
+      },
+      "name": "When a media player is muted"
+    },
     "stopped_playing": {
       "description": "Triggers when a media player stops playing.",
       "description_configured": "[%key:component::media_player::triggers::stopped_playing::description%]",

homeassistant/components/media_player/trigger.py

@@ -1,12 +1,35 @@
 """Provides triggers for media players."""
 
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers.trigger import Trigger, make_conditional_entity_state_trigger
+from homeassistant.core import HomeAssistant, State
+from homeassistant.helpers.trigger import (
+    EntityTriggerBase,
+    Trigger,
+    make_conditional_entity_state_trigger,
+)
 
-from . import MediaPlayerState
+from . import ATTR_MEDIA_VOLUME_LEVEL, ATTR_MEDIA_VOLUME_MUTED, MediaPlayerState
 from .const import DOMAIN
 
+
+class MediaPlayerMutedTrigger(EntityTriggerBase):
+    """Class for media player muted triggers."""
+
+    _domain: str = DOMAIN
+
+    def is_muted(self, state: State) -> bool:
+        """Check if the media player is muted."""
+        return (
+            state.attributes.get(ATTR_MEDIA_VOLUME_MUTED) is True
+            or state.attributes.get(ATTR_MEDIA_VOLUME_LEVEL) == 0
+        )
+
+    def is_to_state(self, state: State) -> bool:
+        """Check if the state matches the target state."""
+        return self.is_muted(state)
+
+
 TRIGGERS: dict[str, type[Trigger]] = {
+    "muted": MediaPlayerMutedTrigger,
     "stopped_playing": make_conditional_entity_state_trigger(
         DOMAIN,
         from_states={

homeassistant/components/media_player/triggers.yaml

@@ -1,4 +1,4 @@
-stopped_playing:
+.trigger_common: &trigger_common
   target:
     entity:
       domain: media_player
@@ -13,3 +13,6 @@ stopped_playing:
             - first
             - last
             - any
+
+muted: *trigger_common
+stopped_playing: *trigger_common

tests/components/media_player/test_trigger.py

@@ -5,8 +5,17 @@ from unittest.mock import patch
 
 import pytest
 
-from homeassistant.components.media_player import MediaPlayerState
-from homeassistant.const import ATTR_LABEL_ID, CONF_ENTITY_ID
+from homeassistant.components.media_player import (
+    ATTR_MEDIA_VOLUME_LEVEL,
+    ATTR_MEDIA_VOLUME_MUTED,
+    MediaPlayerState,
+)
+from homeassistant.const import (
+    ATTR_LABEL_ID,
+    CONF_ENTITY_ID,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+)
 from homeassistant.core import HomeAssistant, ServiceCall
 from homeassistant.setup import async_setup_component
 
@@ -60,6 +69,151 @@ async def test_media_player_triggers_gated_by_labs_flag(
     ) in caplog.text
 
 
+def parametrize_muted_unmuted_trigger_states(
+    trigger: str,
+    target_state: tuple[str, dict],
+    other_state: tuple[str, dict],
+) -> list[
+    tuple[str, tuple[str | None, dict], list[tuple[tuple[str | None, dict], int]]]
+]:
+    """Parametrize states and expected service call counts.
+
+    Returns a list of tuples with (trigger, initial_state, list of states), where
+    states is a list of tuples
+    (state to set, expected service call count).
+    """
+    return [
+        # Initial state None
+        (
+            trigger,
+            (None, {}),
+            [
+                (target_state, 0),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+        # Initial state different from target state
+        (
+            trigger,
+            other_state,
+            [
+                (target_state, 1),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+        # Initial state same as target state
+        (
+            trigger,
+            target_state,
+            [
+                (target_state, 0),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+        # Initial state unavailable / unknown
+        (
+            trigger,
+            (STATE_UNAVAILABLE, {}),
+            [
+                (target_state, 0),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+        (
+            trigger,
+            (STATE_UNKNOWN, {}),
+            [
+                (target_state, 0),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+    ]
+
+
+def parametrize_muted_trigger_states() -> list[
+    tuple[str, tuple[str | None, dict], list[tuple[tuple[str | None, dict], int]]]
+]:
+    """Parametrize states and expected service call counts.
+
+    Returns a list of tuples with (trigger, initial_state, list of states), where
+    states is a list of tuples (state to set, expected service call count).
+    """
+    trigger = "media_player.muted"
+    return [
+        # States with muted attribute
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: True}),
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: False}),
+        ),
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: True}),
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: None}),
+        ),
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: True}),
+            (MediaPlayerState.PLAYING, {}),  # Missing attribute
+        ),
+        # States with volume attribute
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 0}),
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 1}),
+        ),
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 0}),
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: None}),
+        ),
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 0}),
+            (MediaPlayerState.PLAYING, {}),  # Missing attribute
+        ),
+        # States with muted and volume attribute
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 0, ATTR_MEDIA_VOLUME_MUTED: True},
+            ),
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: False},
+            ),
+        ),
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 0, ATTR_MEDIA_VOLUME_MUTED: False},
+            ),
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: False},
+            ),
+        ),
+        *parametrize_muted_unmuted_trigger_states(
+            trigger,
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: True},
+            ),
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: False},
+            ),
+        ),
+    ]
+
+
 @pytest.mark.usefixtures("enable_experimental_triggers_conditions")
 @pytest.mark.parametrize(
     ("trigger_target_config", "entity_id", "entities_in_target"),
@@ -121,6 +275,56 @@ async def test_media_player_state_trigger_behavior_any(
         service_calls.clear()
 
 
+@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "initial_state", "states"),
+    [
+        *parametrize_muted_trigger_states(),
+    ],
+)
+async def test_media_player_state_attribute_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_media_players: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the media player state trigger fires when any media player state changes to a specific state."""
+    await async_setup_component(hass, "media player", {})
+
+    other_entity_ids = set(target_media_players) - {entity_id}
+
+    # Set all media players, including the tested media player, to the initial state
+    for eid in target_media_players:
+        set_or_remove_state(hass, eid, initial_state[0], initial_state[1])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states:
+        set_or_remove_state(hass, entity_id, state[0], state[1])
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other media players also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state[0], state[1])
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
 @pytest.mark.usefixtures("enable_experimental_triggers_conditions")
 @pytest.mark.parametrize(
     ("trigger_target_config", "entity_id", "entities_in_target"),
@@ -181,6 +385,60 @@ async def test_media_player_state_trigger_behavior_first(
         assert len(service_calls) == 0
 
 
+@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "initial_state", "states"),
+    [
+        *parametrize_muted_trigger_states(),
+    ],
+)
+async def test_media_player_state_attribute_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_media_players: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the media player state trigger fires when the first media player state changes to a specific state."""
+    await async_setup_component(hass, "media_player", {})
+
+    other_entity_ids = set(target_media_players) - {entity_id}
+
+    # Set all media players, including the tested media player, to the initial state
+    for eid in target_media_players:
+        set_or_remove_state(hass, eid, initial_state[0], initial_state[1])
+        await hass.async_block_till_done()
+
+    await arm_trigger(
+        hass,
+        trigger,
+        {"behavior": "first"},
+        trigger_target_config,
+    )
+
+    for state, expected_calls in states:
+        set_or_remove_state(hass, entity_id, state[0], state[1])
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other media players should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state[0], state[1])
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
 @pytest.mark.usefixtures("enable_experimental_triggers_conditions")
 @pytest.mark.parametrize(
     ("trigger_target_config", "entity_id", "entities_in_target"),
@@ -238,3 +496,51 @@ async def test_media_player_state_trigger_behavior_last(
         for service_call in service_calls:
             assert service_call.data[CONF_ENTITY_ID] == entity_id
         service_calls.clear()
+
+
+@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "initial_state", "states"),
+    [
+        *parametrize_muted_trigger_states(),
+    ],
+)
+async def test_media_player_state_attribute_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_media_players: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the media player state trigger fires when the last media player state changes to a specific state."""
+    await async_setup_component(hass, "media_player", {})
+
+    other_entity_ids = set(target_media_players) - {entity_id}
+
+    # Set all media players, including the tested media player, to the initial state
+    for eid in target_media_players:
+        set_or_remove_state(hass, eid, initial_state[0], initial_state[1])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state[0], state[1])
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, state[0], state[1])
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()