core/homeassistant/components/growatt_server/switch.py

"""Switch platform for Growatt."""

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, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity

from .const import DOMAIN
from .coordinator import GrowattConfigEntry, GrowattCoordinator

_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):
    """Describes Growatt switch entity."""

    api_key: str
    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
    ),
)


def _create_switches_for_device(
    coordinator: GrowattCoordinator,
) -> list[GrowattSwitch]:
    """Create switch entities for a device coordinator."""
    if coordinator.device_type == "min" and coordinator.api_version == "v1":
        return [
            GrowattSwitch(coordinator, description) for description in MIN_SWITCH_TYPES
        ]
    return []


async def async_setup_entry(
    hass: HomeAssistant,
    entry: GrowattConfigEntry,
    async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
    """Set up Growatt switch entities."""
    runtime_data = entry.runtime_data

    async_add_entities(
        entity
        for coordinator in runtime_data.devices.values()
        for entity in _create_switches_for_device(coordinator)
    )

    @callback
    def _async_new_device(coordinators: list[GrowattCoordinator]) -> None:
        """Add switch entities for new devices."""
        new_entities = [
            entity
            for coordinator in coordinators
            for entity in _create_switches_for_device(coordinator)
        ]
        if new_entities:
            async_add_entities(new_entities)

    entry.async_on_unload(
        async_dispatcher_connect(
            hass,
            f"{DOMAIN}_new_device_{entry.entry_id}",
            _async_new_device,
        )
    )


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,
            serial_number=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(
                translation_domain=DOMAIN,
                translation_key="api_error",
                translation_placeholders={"error": str(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()