Add Powerwall off grid switch (#86357)

Co-authored-by: J. Nick Koston <nick@koston.org>

CODEOWNERS

@@ -894,8 +894,8 @@ build.json @home-assistant/supervisor
 /tests/components/point/ @fredrike
 /homeassistant/components/poolsense/ @haemishkyd
 /tests/components/poolsense/ @haemishkyd
-/homeassistant/components/powerwall/ @bdraco @jrester
-/tests/components/powerwall/ @bdraco @jrester
+/homeassistant/components/powerwall/ @bdraco @jrester @daniel-simpson
+/tests/components/powerwall/ @bdraco @jrester @daniel-simpson
 /homeassistant/components/profiler/ @bdraco
 /tests/components/profiler/ @bdraco
 /homeassistant/components/progettihwsw/ @ardaseremet

homeassistant/components/powerwall/__init__.py

@@ -35,7 +35,7 @@
 
 CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
 
-PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
+PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR, Platform.SWITCH]
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -156,6 +156,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         base_info=base_info,
         http_session=http_session,
         coordinator=None,
+        api_instance=power_wall,
     )
 
     manager = PowerwallDataManager(hass, power_wall, ip_address, password, runtime_data)

homeassistant/components/powerwall/binary_sensor.py

@@ -14,6 +14,11 @@
 from .entity import PowerWallEntity
 from .models import PowerwallRuntimeData
 
+CONNECTED_GRID_STATUSES = {
+    GridStatus.TRANSITION_TO_GRID,
+    GridStatus.CONNECTED,
+}
+
 
 async def async_setup_entry(
     hass: HomeAssistant,
@@ -101,7 +106,7 @@ def unique_id(self) -> str:
     @property
     def is_on(self) -> bool:
         """Grid is online."""
-        return self.data.grid_status == GridStatus.CONNECTED
+        return self.data.grid_status in CONNECTED_GRID_STATUSES
 
 
 class PowerWallChargingStatusSensor(PowerWallEntity, BinarySensorEntity):

homeassistant/components/powerwall/const.py

@@ -5,6 +5,7 @@
 
 POWERWALL_BASE_INFO: Final = "base_info"
 POWERWALL_COORDINATOR: Final = "coordinator"
+POWERWALL_API: Final = "api_instance"
 POWERWALL_API_CHANGED: Final = "api_changed"
 POWERWALL_HTTP_SESSION: Final = "http_session"
 

homeassistant/components/powerwall/entity.py

@@ -10,6 +10,7 @@
     DOMAIN,
     MANUFACTURER,
     MODEL,
+    POWERWALL_API,
     POWERWALL_BASE_INFO,
     POWERWALL_COORDINATOR,
 )
@@ -25,6 +26,7 @@ def __init__(self, powerwall_data: PowerwallRuntimeData) -> None:
         coordinator = powerwall_data[POWERWALL_COORDINATOR]
         assert coordinator is not None
         super().__init__(coordinator)
+        self.power_wall = powerwall_data[POWERWALL_API]
         # The serial numbers of the powerwalls are unique to every site
         self.base_unique_id = "_".join(base_info.serial_numbers)
         self._attr_device_info = DeviceInfo(

homeassistant/components/powerwall/manifest.json

@@ -4,7 +4,7 @@
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/powerwall",
   "requirements": ["tesla-powerwall==0.3.19"],
-  "codeowners": ["@bdraco", "@jrester"],
+  "codeowners": ["@bdraco", "@jrester", "@daniel-simpson"],
   "dhcp": [
     {
       "hostname": "1118431-*"

homeassistant/components/powerwall/models.py

@@ -9,6 +9,7 @@
     DeviceType,
     GridStatus,
     MetersAggregates,
+    Powerwall,
     PowerwallStatus,
     SiteInfo,
     SiteMaster,
@@ -45,6 +46,7 @@ class PowerwallRuntimeData(TypedDict):
     """Run time data for the powerwall."""
 
     coordinator: DataUpdateCoordinator[PowerwallData] | None
+    api_instance: Powerwall
     base_info: PowerwallBaseInfo
     api_changed: bool
     http_session: Session

homeassistant/components/powerwall/switch.py

@@ -0,0 +1,74 @@
+"""Support for Powerwall Switches (V2 API only)."""
+
+from typing import Any
+
+from tesla_powerwall import GridStatus, IslandMode, PowerwallError
+
+from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity import EntityCategory
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN
+from .entity import PowerWallEntity
+from .models import PowerwallRuntimeData
+
+OFF_GRID_STATUSES = {
+    GridStatus.TRANSITION_TO_ISLAND,
+    GridStatus.ISLANDED,
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Powerwall switch platform from Powerwall resources."""
+    powerwall_data: PowerwallRuntimeData = hass.data[DOMAIN][config_entry.entry_id]
+    async_add_entities([PowerwallOffGridEnabledEntity(powerwall_data)])
+
+
+class PowerwallOffGridEnabledEntity(PowerWallEntity, SwitchEntity):
+    """Representation of a Switch entity for Powerwall Off-grid operation."""
+
+    _attr_name = "Off-Grid operation"
+    _attr_has_entity_name = True
+    _attr_entity_category = EntityCategory.CONFIG
+    _attr_device_class = SwitchDeviceClass.SWITCH
+
+    def __init__(self, powerwall_data: PowerwallRuntimeData) -> None:
+        """Initialize powerwall entity and unique id."""
+        super().__init__(powerwall_data)
+        self._attr_unique_id = f"{self.base_unique_id}_off_grid_operation"
+
+    @property
+    def is_on(self) -> bool:
+        """Return true if the powerwall is off-grid."""
+        return self.coordinator.data.grid_status in OFF_GRID_STATUSES
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn off-grid mode on."""
+        await self._async_set_island_mode(IslandMode.OFFGRID)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn off-grid mode off (return to on-grid usage)."""
+        await self._async_set_island_mode(IslandMode.ONGRID)
+
+    async def _async_set_island_mode(self, island_mode: IslandMode) -> None:
+        """Toggles off-grid mode using the island_mode argument."""
+        try:
+            await self.hass.async_add_executor_job(
+                self.power_wall.set_island_mode, island_mode
+            )
+        except PowerwallError as ex:
+            raise HomeAssistantError(
+                f"Setting off-grid operation to {island_mode} failed: {ex}"
+            ) from ex
+
+        self._attr_is_on = island_mode == IslandMode.OFFGRID
+        self.async_write_ha_state()
+
+        await self.coordinator.async_request_refresh()

tests/components/powerwall/test_switch.py

@@ -0,0 +1,104 @@
+"""Test for Powerwall off-grid switch."""
+
+from unittest.mock import Mock, patch
+
+import pytest
+from tesla_powerwall import GridStatus, PowerwallError
+
+from homeassistant.components.powerwall.const import DOMAIN
+from homeassistant.components.switch import (
+    DOMAIN as SWITCH_DOMAIN,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+)
+from homeassistant.const import ATTR_ENTITY_ID, CONF_IP_ADDRESS, STATE_OFF, STATE_ON
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as ent_reg
+
+from .mocks import _mock_powerwall_with_fixtures
+
+from tests.common import MockConfigEntry
+
+ENTITY_ID = "switch.mysite_off_grid_operation"
+
+
+@pytest.fixture(name="mock_powerwall")
+async def mock_powerwall_fixture(hass):
+    """Set up base powerwall fixture."""
+
+    mock_powerwall = await _mock_powerwall_with_fixtures(hass)
+
+    config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_IP_ADDRESS: "1.2.3.4"})
+    config_entry.add_to_hass(hass)
+    with patch(
+        "homeassistant.components.powerwall.Powerwall", return_value=mock_powerwall
+    ):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+        yield mock_powerwall
+
+
+async def test_entity_registry(hass, mock_powerwall):
+    """Test powerwall off-grid switch device."""
+
+    mock_powerwall.get_grid_status = Mock(return_value=GridStatus.CONNECTED)
+    entity_registry = ent_reg.async_get(hass)
+
+    assert ENTITY_ID in entity_registry.entities
+
+
+async def test_initial(hass, mock_powerwall):
+    """Test initial grid status without off grid switch selected."""
+
+    mock_powerwall.get_grid_status = Mock(return_value=GridStatus.CONNECTED)
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_OFF
+
+
+async def test_on(hass, mock_powerwall):
+    """Test state once offgrid switch has been turned on."""
+
+    mock_powerwall.get_grid_status = Mock(return_value=GridStatus.ISLANDED)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        SERVICE_TURN_ON,
+        {ATTR_ENTITY_ID: ENTITY_ID},
+        blocking=True,
+    )
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_ON
+
+
+async def test_off(hass, mock_powerwall):
+    """Test state once offgrid switch has been turned off."""
+
+    mock_powerwall.get_grid_status = Mock(return_value=GridStatus.CONNECTED)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        SERVICE_TURN_OFF,
+        {ATTR_ENTITY_ID: ENTITY_ID},
+        blocking=True,
+    )
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_OFF
+
+
+async def test_exception_on_powerwall_error(hass, mock_powerwall):
+    """Ensure that an exception in the tesla_powerwall library causes a HomeAssistantError."""
+
+    with pytest.raises(HomeAssistantError, match="Setting off-grid operation to"):
+        mock_powerwall.set_island_mode = Mock(
+            side_effect=PowerwallError("Mock exception")
+        )
+
+        await hass.services.async_call(
+            SWITCH_DOMAIN,
+            SERVICE_TURN_ON,
+            {ATTR_ENTITY_ID: ENTITY_ID},
+            blocking=True,
+        )