1
0
mirror of https://github.com/home-assistant/core.git synced 2025-12-20 02:48:57 +00:00

Add switch platform and grid charge enable for Growatt Server integration (#153960)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
johanzander
2025-10-13 17:54:30 +02:00
committed by GitHub
parent 1452aec47f
commit dd6bc715d8
6 changed files with 703 additions and 1 deletions

View File

@@ -36,7 +36,7 @@ DEFAULT_URL = SERVER_URLS[0]
DOMAIN = "growatt_server"
PLATFORMS = [Platform.SENSOR]
PLATFORMS = [Platform.SENSOR, Platform.SWITCH]
LOGIN_INVALID_AUTH_CODE = "502"

View File

@@ -461,6 +461,11 @@
"total_maximum_output": {
"name": "Maximum power"
}
},
"switch": {
"ac_charge": {
"name": "Charge from grid"
}
}
}
}

View File

@@ -0,0 +1,138 @@
"""Switch platform for Growatt."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import Any
from growattServer import GrowattV1ApiError
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import GrowattConfigEntry, GrowattCoordinator
from .sensor.sensor_entity_description import GrowattRequiredKeysMixin
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = (
1 # Serialize updates as inverter does not handle concurrent requests
)
@dataclass(frozen=True, kw_only=True)
class GrowattSwitchEntityDescription(SwitchEntityDescription, GrowattRequiredKeysMixin):
"""Describes Growatt switch entity."""
write_key: str | None = None # Parameter ID for writing (if different from api_key)
# Note that the Growatt V1 API uses different keys for reading and writing parameters.
# Reading values returns camelCase keys, while writing requires snake_case keys.
MIN_SWITCH_TYPES: tuple[GrowattSwitchEntityDescription, ...] = (
GrowattSwitchEntityDescription(
key="ac_charge",
translation_key="ac_charge",
api_key="acChargeEnable", # Key returned by V1 API
write_key="ac_charge", # Key used to write parameter
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: GrowattConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Growatt switch entities."""
runtime_data = entry.runtime_data
# Add switch entities for each MIN device (only supported with V1 API)
async_add_entities(
GrowattSwitch(device_coordinator, description)
for device_coordinator in runtime_data.devices.values()
if (
device_coordinator.device_type == "min"
and device_coordinator.api_version == "v1"
)
for description in MIN_SWITCH_TYPES
)
class GrowattSwitch(CoordinatorEntity[GrowattCoordinator], SwitchEntity):
"""Representation of a Growatt switch."""
_attr_has_entity_name = True
_attr_entity_category = EntityCategory.CONFIG
entity_description: GrowattSwitchEntityDescription
def __init__(
self,
coordinator: GrowattCoordinator,
description: GrowattSwitchEntityDescription,
) -> None:
"""Initialize the switch."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.device_id}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.device_id)},
manufacturer="Growatt",
name=coordinator.device_id,
)
@property
def is_on(self) -> bool | None:
"""Return true if the switch is on."""
value = self.coordinator.data.get(self.entity_description.api_key)
if value is None:
return None
# API returns integer 1 for enabled, 0 for disabled
return bool(value)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self._async_set_state(True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self._async_set_state(False)
async def _async_set_state(self, state: bool) -> None:
"""Set the switch state."""
# Use write_key if specified, otherwise fall back to api_key
parameter_id = (
self.entity_description.write_key or self.entity_description.api_key
)
api_value = int(state)
try:
# Use V1 API to write parameter
await self.hass.async_add_executor_job(
self.coordinator.api.min_write_parameter,
self.coordinator.device_id,
parameter_id,
api_value,
)
except GrowattV1ApiError as e:
raise HomeAssistantError(f"Error while setting switch state: {e}") from e
# If no exception was raised, the write was successful
_LOGGER.debug(
"Set switch %s to %s",
parameter_id,
api_value,
)
# Update the value in coordinator data (keep as integer like API returns)
self.coordinator.data[self.entity_description.api_key] = api_value
self.async_write_ha_state()

View File

@@ -0,0 +1,202 @@
"""Common fixtures for the Growatt server tests."""
from unittest.mock import patch
import pytest
from homeassistant.components.growatt_server.const import (
AUTH_API_TOKEN,
AUTH_PASSWORD,
CONF_AUTH_TYPE,
CONF_PLANT_ID,
DEFAULT_URL,
DOMAIN,
)
from homeassistant.const import CONF_PASSWORD, CONF_TOKEN, CONF_URL, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.fixture
def mock_growatt_v1_api():
"""Return a mocked Growatt V1 API.
This fixture provides the happy path for integration setup and basic operations.
Individual tests can override specific return values to test error conditions.
Methods mocked for integration setup:
- device_list: Called during async_setup_entry to discover devices
- plant_energy_overview: Called by total coordinator during first refresh
Methods mocked for MIN device coordinator refresh:
- min_detail: Provides device state (e.g., acChargeEnable for switches)
- min_settings: Provides settings (e.g. TOU periods)
- min_energy: Provides energy data (empty for switch tests, sensors need real data)
Methods mocked for switch operations:
- min_write_parameter: Called by switch entities to change settings
"""
with patch("growattServer.OpenApiV1", autospec=True) as mock_v1_api_class:
mock_v1_api = mock_v1_api_class.return_value
# Called during setup to discover devices
mock_v1_api.device_list.return_value = {
"devices": [
{
"device_sn": "MIN123456",
"type": 7, # MIN device type
}
]
}
# Called by MIN device coordinator during refresh
mock_v1_api.min_detail.return_value = {
"deviceSn": "MIN123456",
"acChargeEnable": 1, # AC charge enabled - read by switch entity
}
# Called by MIN device coordinator during refresh
mock_v1_api.min_settings.return_value = {
# Forced charge time segments (not used by switch, but coordinator fetches it)
"forcedTimeStart1": "06:00",
"forcedTimeStop1": "08:00",
"forcedChargeBatMode1": 1,
"forcedChargeFlag1": 1,
"forcedTimeStart2": "22:00",
"forcedTimeStop2": "24:00",
"forcedChargeBatMode2": 0,
"forcedChargeFlag2": 0,
}
# Called by MIN device coordinator during refresh
# Empty dict is sufficient for switch tests (sensor tests would need real energy data)
mock_v1_api.min_energy.return_value = {}
# Called by total coordinator during refresh
mock_v1_api.plant_energy_overview.return_value = {
"today_energy": 12.5,
"total_energy": 1250.0,
"current_power": 2500,
}
# Called by switch entities during turn_on/turn_off
mock_v1_api.min_write_parameter.return_value = None
yield mock_v1_api
@pytest.fixture
def mock_growatt_classic_api():
"""Return a mocked Growatt Classic API.
This fixture provides the happy path for Classic API integration setup.
Individual tests can override specific return values to test error conditions.
Methods mocked for integration setup:
- login: Called during get_device_list_classic to authenticate
- plant_list: Called during setup if plant_id is default (to auto-select plant)
- device_list: Called during async_setup_entry to discover devices
Methods mocked for total coordinator refresh:
- plant_info: Provides plant totals (energy, power, money) for Classic API
Methods mocked for device-specific tests:
- tlx_detail: Provides TLX device data (kept for potential future tests)
"""
with patch("growattServer.GrowattApi", autospec=True) as mock_classic_api_class:
# Use the autospec'd mock instance instead of creating a new Mock()
mock_classic_api = mock_classic_api_class.return_value
# Called during setup to authenticate with Classic API
mock_classic_api.login.return_value = {"success": True, "user": {"id": 12345}}
# Called during setup if plant_id is default (auto-select first plant)
mock_classic_api.plant_list.return_value = {"data": [{"plantId": "12345"}]}
# Called during setup to discover devices
mock_classic_api.device_list.return_value = [
{"deviceSn": "MIN123456", "deviceType": "min"}
]
# Called by total coordinator during refresh for Classic API
mock_classic_api.plant_info.return_value = {
"deviceList": [],
"totalEnergy": 1250.0,
"todayEnergy": 12.5,
"invTodayPpv": 2500,
"plantMoneyText": "123.45/USD",
}
# Called for TLX device coordinator (kept for potential future tests)
mock_classic_api.tlx_detail.return_value = {
"data": {
"deviceSn": "TLX123456",
}
}
yield mock_classic_api
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry (V1 API with token auth).
This is the primary config entry used by most tests. For Classic API tests,
use mock_config_entry_classic instead.
"""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_AUTH_TYPE: AUTH_API_TOKEN,
CONF_TOKEN: "test_token_123",
CONF_URL: DEFAULT_URL,
"user_id": "12345",
CONF_PLANT_ID: "plant_123",
"name": "Test Plant",
},
unique_id="plant_123",
)
@pytest.fixture
def mock_config_entry_classic() -> MockConfigEntry:
"""Return a mocked config entry for Classic API (password auth).
Use this for tests that specifically need to test Classic API behavior.
Most tests use the default mock_config_entry (V1 API) instead.
"""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_AUTH_TYPE: AUTH_PASSWORD,
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
CONF_URL: DEFAULT_URL,
CONF_PLANT_ID: "12345",
"name": "Test Plant",
},
unique_id="12345",
)
@pytest.fixture
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_growatt_v1_api
) -> MockConfigEntry:
"""Set up the Growatt Server integration for testing (V1 API).
This combines mock_config_entry and mock_growatt_v1_api to provide a fully
initialized integration ready for testing. Use @pytest.mark.usefixtures("init_integration")
to automatically set up the integration before your test runs.
For Classic API tests, manually set up using mock_config_entry_classic and
mock_growatt_classic_api instead.
"""
# The mock_growatt_v1_api fixture is required for patches to be active
assert mock_growatt_v1_api is not None
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

View File

@@ -0,0 +1,80 @@
# serializer version: 1
# name: test_switch_device_registry
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'growatt_server',
'MIN123456',
),
}),
'labels': set({
}),
'manufacturer': 'Growatt',
'model': None,
'model_id': None,
'name': 'MIN123456',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_switch_entities[switch.min123456_charge_from_grid-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': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.min123456_charge_from_grid',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Charge from grid',
'platform': 'growatt_server',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ac_charge',
'unique_id': 'MIN123456_ac_charge',
'unit_of_measurement': None,
})
# ---
# name: test_switch_entities[switch.min123456_charge_from_grid-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'MIN123456 Charge from grid',
}),
'context': <ANY>,
'entity_id': 'switch.min123456_charge_from_grid',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

View File

@@ -0,0 +1,277 @@
"""Tests for the Growatt Server switch platform."""
from collections.abc import AsyncGenerator
from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
from growattServer import GrowattV1ApiError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.growatt_server.coordinator import SCAN_INTERVAL
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNKNOWN,
EntityCategory,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
DOMAIN = "growatt_server"
@pytest.fixture(autouse=True)
async def switch_only() -> AsyncGenerator[None]:
"""Enable only the switch platform."""
with patch(
"homeassistant.components.growatt_server.PLATFORMS",
[Platform.SWITCH],
):
yield
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_switch_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that switch entities are created for MIN devices."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_turn_on_switch_success(
hass: HomeAssistant,
mock_growatt_v1_api,
) -> None:
"""Test turning on a switch entity successfully."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.min123456_charge_from_grid"},
blocking=True,
)
# Verify API was called with correct parameters
mock_growatt_v1_api.min_write_parameter.assert_called_once_with(
"MIN123456", "ac_charge", 1
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_turn_off_switch_success(
hass: HomeAssistant,
mock_growatt_v1_api,
) -> None:
"""Test turning off a switch entity successfully."""
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.min123456_charge_from_grid"},
blocking=True,
)
# Verify API was called with correct parameters
mock_growatt_v1_api.min_write_parameter.assert_called_once_with(
"MIN123456", "ac_charge", 0
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_turn_on_switch_api_error(
hass: HomeAssistant,
mock_growatt_v1_api,
) -> None:
"""Test handling API error when turning on switch."""
# Mock API to raise error
mock_growatt_v1_api.min_write_parameter.side_effect = GrowattV1ApiError("API Error")
with pytest.raises(HomeAssistantError, match="Error while setting switch state"):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{"entity_id": "switch.min123456_charge_from_grid"},
blocking=True,
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_turn_off_switch_api_error(
hass: HomeAssistant,
mock_growatt_v1_api,
) -> None:
"""Test handling API error when turning off switch."""
# Mock API to raise error
mock_growatt_v1_api.min_write_parameter.side_effect = GrowattV1ApiError("API Error")
with pytest.raises(HomeAssistantError, match="Error while setting switch state"):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{"entity_id": "switch.min123456_charge_from_grid"},
blocking=True,
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_switch_entity_attributes(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Test switch entity attributes."""
# Check entity registry attributes
entity_entry = entity_registry.async_get("switch.min123456_charge_from_grid")
assert entity_entry is not None
assert entity_entry.entity_category == EntityCategory.CONFIG
assert entity_entry.unique_id == "MIN123456_ac_charge"
# Check state attributes
state = hass.states.get("switch.min123456_charge_from_grid")
assert state is not None
assert state.attributes["friendly_name"] == "MIN123456 Charge from grid"
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
async def test_switch_device_registry(
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test that switch entities are associated with the correct device."""
# Get the device from device registry
device = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
assert device is not None
assert device == snapshot
# Verify switch entity is associated with the device
entity_entry = entity_registry.async_get("switch.min123456_charge_from_grid")
assert entity_entry is not None
assert entity_entry.device_id == device.id
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_switch_state_handling_integer_values(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_growatt_v1_api,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test switch state handling with integer values from API."""
# Set up integration
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Should interpret 1 as ON (from default mock data)
state = hass.states.get("switch.min123456_charge_from_grid")
assert state is not None
assert state.state == STATE_ON
# Test with 0 integer value
mock_growatt_v1_api.min_detail.return_value = {
"deviceSn": "MIN123456",
"acChargeEnable": 0, # Integer value
}
# Advance time to trigger coordinator refresh
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
# Should interpret 0 as OFF
state = hass.states.get("switch.min123456_charge_from_grid")
assert state is not None
assert state.state == STATE_OFF
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_switch_missing_data(
hass: HomeAssistant,
mock_growatt_v1_api,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test switch entity when coordinator data is missing."""
# Set up API with missing data for switch entity
mock_growatt_v1_api.min_detail.return_value = {
"deviceSn": "MIN123456",
# Missing 'acChargeEnable' key to test None case
}
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Entity should exist but have unknown state due to missing data
state = hass.states.get("switch.min123456_charge_from_grid")
assert state is not None
assert state.state == STATE_UNKNOWN
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_no_switch_entities_for_non_min_devices(
hass: HomeAssistant,
mock_growatt_v1_api,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that switch entities are not created for non-MIN devices."""
# Mock a different device type (not MIN) - type 7 is MIN, type 8 is non-MIN
mock_growatt_v1_api.device_list.return_value = {
"devices": [
{
"device_sn": "TLX123456",
"type": 8, # Non-MIN device type (MIN is type 7)
}
]
}
# Mock TLX API response to prevent coordinator errors
mock_growatt_v1_api.tlx_detail.return_value = {"data": {}}
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Should have no switch entities for TLX devices
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
switch_entities = [entry for entry in entity_entries if entry.domain == "switch"]
assert len(switch_entities) == 0
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_no_switch_entities_for_classic_api(
hass: HomeAssistant,
mock_growatt_classic_api,
mock_config_entry_classic: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that switch entities are not created for Classic API."""
# Mock device list to return no devices
mock_growatt_classic_api.device_list.return_value = []
mock_config_entry_classic.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry_classic.entry_id)
await hass.async_block_till_done()
# Should have no switch entities for classic API (no devices)
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry_classic.entry_id
)
switch_entities = [entry for entry in entity_entries if entry.domain == "switch"]
assert len(switch_entities) == 0