Add button platform to Indevolt integration (#165283)

homeassistant/components/indevolt/__init__.py

@@ -8,6 +8,7 @@ from homeassistant.core import HomeAssistant
 from .coordinator import IndevoltConfigEntry, IndevoltCoordinator
 
 PLATFORMS: list[Platform] = [
+    Platform.BUTTON,
     Platform.NUMBER,
     Platform.SELECT,
     Platform.SENSOR,

homeassistant/components/indevolt/button.py

@@ -0,0 +1,70 @@
+"""Button platform for Indevolt integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Final
+
+from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from . import IndevoltConfigEntry
+from .coordinator import IndevoltCoordinator
+from .entity import IndevoltEntity
+
+PARALLEL_UPDATES = 0
+
+
+@dataclass(frozen=True, kw_only=True)
+class IndevoltButtonEntityDescription(ButtonEntityDescription):
+    """Custom entity description class for Indevolt button entities."""
+
+    generation: list[int] = field(default_factory=lambda: [1, 2])
+
+
+BUTTONS: Final = (
+    IndevoltButtonEntityDescription(
+        key="stop",
+        translation_key="stop",
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: IndevoltConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the button platform for Indevolt."""
+    coordinator = entry.runtime_data
+    device_gen = coordinator.generation
+
+    # Button initialization
+    async_add_entities(
+        IndevoltButtonEntity(coordinator=coordinator, description=description)
+        for description in BUTTONS
+        if device_gen in description.generation
+    )
+
+
+class IndevoltButtonEntity(IndevoltEntity, ButtonEntity):
+    """Represents a button entity for Indevolt devices."""
+
+    entity_description: IndevoltButtonEntityDescription
+
+    def __init__(
+        self,
+        coordinator: IndevoltCoordinator,
+        description: IndevoltButtonEntityDescription,
+    ) -> None:
+        """Initialize the Indevolt button entity."""
+        super().__init__(coordinator)
+
+        self.entity_description = description
+        self._attr_unique_id = f"{self.serial_number}_{description.key}"
+
+    async def async_press(self) -> None:
+        """Handle the button press."""
+
+        await self.coordinator.async_execute_realtime_action([0, 0, 0])

homeassistant/components/indevolt/const.py

@@ -1,16 +1,27 @@
 """Constants for the Indevolt integration."""
 
-DOMAIN = "indevolt"
+from typing import Final
+
+DOMAIN: Final = "indevolt"
+
+# Default configurations
+DEFAULT_PORT: Final = 8080
 
 # Config entry fields
-CONF_SERIAL_NUMBER = "serial_number"
-CONF_GENERATION = "generation"
+CONF_SERIAL_NUMBER: Final = "serial_number"
+CONF_GENERATION: Final = "generation"
 
-# Default values
-DEFAULT_PORT = 8080
+# API write/read keys for energy and value for outdoor/portable mode
+ENERGY_MODE_READ_KEY: Final = "7101"
+ENERGY_MODE_WRITE_KEY: Final = "47005"
+PORTABLE_MODE: Final = 0
+
+# API write key and value for real-time control mode
+REALTIME_ACTION_KEY: Final = "47015"
+REALTIME_ACTION_MODE: Final = 4
 
 # API key fields
-SENSOR_KEYS = {
+SENSOR_KEYS: Final[dict[int, list[str]]] = {
     1: [
         "606",
         "7101",

homeassistant/components/indevolt/coordinator.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from datetime import timedelta
 import logging
-from typing import Any
+from typing import Any, Final
 
 from aiohttp import ClientError
 from indevolt_api import IndevoltAPI, TimeOutException
@@ -21,20 +21,37 @@ from .const import (
     CONF_SERIAL_NUMBER,
     DEFAULT_PORT,
     DOMAIN,
+    ENERGY_MODE_READ_KEY,
+    ENERGY_MODE_WRITE_KEY,
+    PORTABLE_MODE,
+    REALTIME_ACTION_KEY,
+    REALTIME_ACTION_MODE,
     SENSOR_KEYS,
 )
 
 _LOGGER = logging.getLogger(__name__)
-SCAN_INTERVAL = 30
+SCAN_INTERVAL: Final = 30
 
 type IndevoltConfigEntry = ConfigEntry[IndevoltCoordinator]
 
 
+class DeviceTimeoutError(HomeAssistantError):
+    """Raised when device push times out."""
+
+
+class DeviceConnectionError(HomeAssistantError):
+    """Raised when device push fails due to connection issues."""
+
+
 class IndevoltCoordinator(DataUpdateCoordinator[dict[str, Any]]):
     """Coordinator for fetching and pushing data to indevolt devices."""
 
+    friendly_name: str
     config_entry: IndevoltConfigEntry
     firmware_version: str | None
+    serial_number: str
+    device_model: str
+    generation: int
 
     def __init__(self, hass: HomeAssistant, entry: IndevoltConfigEntry) -> None:
         """Initialize the indevolt coordinator."""
@@ -53,6 +70,7 @@ class IndevoltCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             session=async_get_clientsession(hass),
         )
 
+        self.friendly_name = entry.title
         self.serial_number = entry.data[CONF_SERIAL_NUMBER]
         self.device_model = entry.data[CONF_MODEL]
         self.generation = entry.data[CONF_GENERATION]
@@ -85,6 +103,67 @@ class IndevoltCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         try:
             return await self.api.set_data(sensor_key, value)
         except TimeOutException as err:
-            raise HomeAssistantError(f"Device push timed out: {err}") from err
+            raise DeviceTimeoutError(f"Device push timed out: {err}") from err
         except (ClientError, ConnectionError, OSError) as err:
-            raise HomeAssistantError(f"Device push failed: {err}") from err
+            raise DeviceConnectionError(f"Device push failed: {err}") from err
+
+    async def async_switch_energy_mode(
+        self, target_mode: int, refresh: bool = True
+    ) -> None:
+        """Attempt to switch device to given energy mode."""
+        current_mode = self.data.get(ENERGY_MODE_READ_KEY)
+
+        # Ensure current energy mode is known
+        if current_mode is None:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="failed_to_retrieve_current_energy_mode",
+            )
+
+        # Ensure device is not in "Outdoor/Portable mode"
+        if current_mode == PORTABLE_MODE:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="energy_mode_change_unavailable_outdoor_portable",
+            )
+
+        # Switch energy mode if required
+        if current_mode != target_mode:
+            try:
+                success = await self.async_push_data(ENERGY_MODE_WRITE_KEY, target_mode)
+            except (DeviceTimeoutError, DeviceConnectionError) as err:
+                raise HomeAssistantError(
+                    translation_domain=DOMAIN,
+                    translation_key="failed_to_switch_energy_mode",
+                ) from err
+
+            if not success:
+                raise HomeAssistantError(
+                    translation_domain=DOMAIN,
+                    translation_key="failed_to_switch_energy_mode",
+                )
+
+            if refresh:
+                await self.async_request_refresh()
+
+    async def async_execute_realtime_action(self, action: list[int]) -> None:
+        """Switch mode, execute action, and refresh for real-time control."""
+
+        await self.async_switch_energy_mode(REALTIME_ACTION_MODE, refresh=False)
+
+        try:
+            success = await self.async_push_data(REALTIME_ACTION_KEY, action)
+
+        except (DeviceTimeoutError, DeviceConnectionError) as err:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="failed_to_execute_realtime_action",
+            ) from err
+
+        if not success:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="failed_to_execute_realtime_action",
+            )
+
+        await self.async_request_refresh()

homeassistant/components/indevolt/strings.json

@@ -35,6 +35,11 @@
     }
   },
   "entity": {
+    "button": {
+      "stop": {
+        "name": "Enable standby mode"
+      }
+    },
     "number": {
       "discharge_limit": {
         "name": "Discharge limit"
@@ -289,5 +294,19 @@
         "name": "LED indicator"
       }
     }
+  },
+  "exceptions": {
+    "energy_mode_change_unavailable_outdoor_portable": {
+      "message": "Energy mode cannot be changed when the device is in outdoor/portable mode"
+    },
+    "failed_to_execute_realtime_action": {
+      "message": "Failed to execute real-time action"
+    },
+    "failed_to_retrieve_current_energy_mode": {
+      "message": "Failed to retrieve current energy mode"
+    },
+    "failed_to_switch_energy_mode": {
+      "message": "Failed to switch to requested energy mode"
+    }
   }
 }

tests/components/indevolt/snapshots/test_button.ambr

@@ -0,0 +1,99 @@
+# serializer version: 1
+# name: test_button[1][button.bk1600_enable_standby_mode-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'button',
+    'entity_category': None,
+    'entity_id': 'button.bk1600_enable_standby_mode',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Enable standby mode',
+    'options': dict({
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': 'Enable standby mode',
+    'platform': 'indevolt',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': 'stop',
+    'unique_id': 'BK1600-12345678_stop',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_button[1][button.bk1600_enable_standby_mode-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'BK1600 Enable standby mode',
+    }),
+    'context': <ANY>,
+    'entity_id': 'button.bk1600_enable_standby_mode',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'unknown',
+  })
+# ---
+# name: test_button[2][button.cms_sf2000_enable_standby_mode-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'button',
+    'entity_category': None,
+    'entity_id': 'button.cms_sf2000_enable_standby_mode',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Enable standby mode',
+    'options': dict({
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': 'Enable standby mode',
+    'platform': 'indevolt',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': 'stop',
+    'unique_id': 'SolidFlex2000-87654321_stop',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_button[2][button.cms_sf2000_enable_standby_mode-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'CMS-SF2000 Enable standby mode',
+    }),
+    'context': <ANY>,
+    'entity_id': 'button.cms_sf2000_enable_standby_mode',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'unknown',
+  })
+# ---

tests/components/indevolt/test_button.py

@@ -0,0 +1,158 @@
+"""Tests for the Indevolt button platform."""
+
+from unittest.mock import AsyncMock, call, patch
+
+from indevolt_api import TimeOutException
+import pytest
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
+from homeassistant.components.indevolt.const import (
+    ENERGY_MODE_READ_KEY,
+    ENERGY_MODE_WRITE_KEY,
+    PORTABLE_MODE,
+    REALTIME_ACTION_KEY,
+    REALTIME_ACTION_MODE,
+)
+from homeassistant.const import ATTR_ENTITY_ID, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er
+
+from . import setup_integration
+
+from tests.common import MockConfigEntry, snapshot_platform
+
+ENTITY_ID_GEN2 = "button.cms_sf2000_enable_standby_mode"
+ENTITY_ID_GEN1 = "button.bk1600_enable_standby_mode"
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+@pytest.mark.parametrize("generation", [2, 1], indirect=True)
+async def test_button(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_indevolt: AsyncMock,
+    snapshot: SnapshotAssertion,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test button entity registration and states."""
+    with patch("homeassistant.components.indevolt.PLATFORMS", [Platform.BUTTON]):
+        await setup_integration(hass, mock_config_entry)
+
+    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+
+
+@pytest.mark.parametrize("generation", [2], indirect=True)
+async def test_button_press_standby(
+    hass: HomeAssistant,
+    mock_indevolt: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test pressing the standby button switches to real-time mode and sends standby action."""
+    with patch("homeassistant.components.indevolt.PLATFORMS", [Platform.BUTTON]):
+        await setup_integration(hass, mock_config_entry)
+
+    # Reset mock call count for this iteration
+    mock_indevolt.set_data.reset_mock()
+
+    # Mock call to pause (dis)charging
+    await hass.services.async_call(
+        BUTTON_DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: ENTITY_ID_GEN2},
+        blocking=True,
+    )
+
+    # Verify set_data was called twice with correct parameters
+    assert mock_indevolt.set_data.call_count == 2
+    mock_indevolt.set_data.assert_has_calls(
+        [
+            call(ENERGY_MODE_WRITE_KEY, REALTIME_ACTION_MODE),
+            call(REALTIME_ACTION_KEY, [0, 0, 0]),
+        ]
+    )
+
+
+@pytest.mark.parametrize("generation", [2], indirect=True)
+async def test_button_press_standby_already_in_realtime_mode(
+    hass: HomeAssistant,
+    mock_indevolt: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test pressing standby when already in real-time mode skips the mode switch."""
+
+    # Force real-time control mode
+    mock_indevolt.fetch_data.return_value[ENERGY_MODE_READ_KEY] = REALTIME_ACTION_MODE
+    with patch("homeassistant.components.indevolt.PLATFORMS", [Platform.BUTTON]):
+        await setup_integration(hass, mock_config_entry)
+
+    # Reset mock call count for this iteration
+    mock_indevolt.set_data.reset_mock()
+
+    # Mock call to pause (dis)charging
+    await hass.services.async_call(
+        BUTTON_DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: ENTITY_ID_GEN2},
+        blocking=True,
+    )
+
+    # Verify set_data was called once with correct parameters
+    mock_indevolt.set_data.assert_called_once_with(REALTIME_ACTION_KEY, [0, 0, 0])
+
+
+@pytest.mark.parametrize("generation", [2], indirect=True)
+async def test_button_press_standby_timeout_error(
+    hass: HomeAssistant,
+    mock_indevolt: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test pressing standby raises HomeAssistantError when the device times out."""
+    with patch("homeassistant.components.indevolt.PLATFORMS", [Platform.BUTTON]):
+        await setup_integration(hass, mock_config_entry)
+
+    # Simulate an API push failure
+    mock_indevolt.set_data.side_effect = TimeOutException("Timed out")
+
+    # Mock call to pause (dis)charging
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: ENTITY_ID_GEN2},
+            blocking=True,
+        )
+
+
+@pytest.mark.parametrize("generation", [2], indirect=True)
+async def test_button_press_standby_portable_mode_error(
+    hass: HomeAssistant,
+    mock_indevolt: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test pressing standby raises HomeAssistantError when device is in outdoor/portable mode."""
+
+    # Force outdoor/portable mode
+    mock_indevolt.fetch_data.return_value[ENERGY_MODE_READ_KEY] = PORTABLE_MODE
+    with patch("homeassistant.components.indevolt.PLATFORMS", [Platform.BUTTON]):
+        await setup_integration(hass, mock_config_entry)
+
+    # Reset mock call count for this iteration
+    mock_indevolt.set_data.reset_mock()
+
+    # Mock call to pause (dis)charging
+    with pytest.raises(HomeAssistantError) as exc_info:
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: ENTITY_ID_GEN2},
+            blocking=True,
+        )
+
+    # Verify correct translation key is used for the error and confirm no call was made
+    assert (
+        exc_info.value.translation_key
+        == "energy_mode_change_unavailable_outdoor_portable"
+    )
+    mock_indevolt.set_data.assert_not_called()