Add services for managing Time-of-Use (TOU) schedule for Growatt integration

homeassistant/components/growatt_server/__init__.py

@@ -10,6 +10,8 @@
 from homeassistant.const import CONF_PASSWORD, CONF_TOKEN, CONF_URL, CONF_USERNAME
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryError
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.typing import ConfigType
 
 from .const import (
     AUTH_API_TOKEN,
@@ -19,14 +21,25 @@
     DEFAULT_PLANT_ID,
     DEFAULT_URL,
     DEPRECATED_URLS,
+    DOMAIN,
     LOGIN_INVALID_AUTH_CODE,
     PLATFORMS,
 )
 from .coordinator import GrowattConfigEntry, GrowattCoordinator
 from .models import GrowattRuntimeData
+from .services import async_register_services
 
 _LOGGER = logging.getLogger(__name__)
 
+CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the Growatt Server component."""
+    # Register services
+    await async_register_services(hass)
+    return True
+
 
 def get_device_list_classic(
     api: growattServer.GrowattApi, config: Mapping[str, str]

homeassistant/components/growatt_server/const.py

@@ -46,3 +46,17 @@
 
 # Config flow abort reasons
 ABORT_NO_PLANTS = "no_plants"
+
+# Battery modes for TOU (Time of Use) settings
+BATT_MODE_LOAD_FIRST = 0
+BATT_MODE_BATTERY_FIRST = 1
+BATT_MODE_GRID_FIRST = 2
+
+BATT_MODE_MAP = {
+    "load-first": BATT_MODE_LOAD_FIRST,
+    "0": BATT_MODE_LOAD_FIRST,
+    "battery-first": BATT_MODE_BATTERY_FIRST,
+    "1": BATT_MODE_BATTERY_FIRST,
+    "grid-first": BATT_MODE_GRID_FIRST,
+    "2": BATT_MODE_GRID_FIRST,
+}

homeassistant/components/growatt_server/coordinator.py

@@ -12,10 +12,17 @@
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_PASSWORD, CONF_URL, CONF_USERNAME
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 from homeassistant.util import dt as dt_util
 
-from .const import DEFAULT_URL, DOMAIN
+from .const import (
+    BATT_MODE_BATTERY_FIRST,
+    BATT_MODE_GRID_FIRST,
+    BATT_MODE_LOAD_FIRST,
+    DEFAULT_URL,
+    DOMAIN,
+)
 from .models import GrowattRuntimeData
 
 if TYPE_CHECKING:
@@ -247,3 +254,134 @@ def get_data(
         self.previous_values[variable] = return_value
 
         return return_value
+
+    async def update_time_segment(
+        self, segment_id: int, batt_mode: int, start_time, end_time, enabled: bool
+    ) -> None:
+        """Update an inverter time segment.
+
+        Args:
+            segment_id: Time segment ID (1-9)
+            batt_mode: Battery mode (0=load first, 1=battery first, 2=grid first)
+            start_time: Start time (datetime.time object)
+            end_time: End time (datetime.time object)
+            enabled: Whether the segment is enabled
+        """
+        _LOGGER.debug(
+            "Updating time segment %d for device %s (mode=%d, %s-%s, enabled=%s)",
+            segment_id,
+            self.device_id,
+            batt_mode,
+            start_time,
+            end_time,
+            enabled,
+        )
+
+        if self.api_version != "v1":
+            raise ServiceValidationError(
+                "Updating time segments requires token authentication"
+            )
+
+        try:
+            # Use V1 API for token authentication
+            # The library's _process_response will raise GrowattV1ApiError if error_code != 0
+            await self.hass.async_add_executor_job(
+                self.api.min_write_time_segment,
+                self.device_id,
+                segment_id,
+                batt_mode,
+                start_time,
+                end_time,
+                enabled,
+            )
+        except growattServer.GrowattV1ApiError as err:
+            raise HomeAssistantError(f"API error updating time segment: {err}") from err
+
+        # Update coordinator's cached data without making an API call (avoids rate limit)
+        if self.data:
+            # Update the time segment data in the cache
+            self.data[f"forcedTimeStart{segment_id}"] = start_time.strftime("%H:%M")
+            self.data[f"forcedTimeStop{segment_id}"] = end_time.strftime("%H:%M")
+            self.data[f"time{segment_id}Mode"] = batt_mode
+            self.data[f"forcedStopSwitch{segment_id}"] = 1 if enabled else 0
+
+            # Notify entities of the updated data (no API call)
+            self.async_set_updated_data(self.data)
+
+    async def read_time_segments(self) -> list[dict]:
+        """Read time segments from an inverter.
+
+        Returns:
+            List of dictionaries containing segment information
+        """
+        _LOGGER.debug("Reading time segments for device %s", self.device_id)
+
+        if self.api_version != "v1":
+            raise ServiceValidationError(
+                "Reading time segments requires token authentication"
+            )
+
+        # Ensure we have current data
+        if not self.data:
+            _LOGGER.debug("Coordinator data not available, triggering refresh")
+            await self.async_refresh()
+
+        time_segments = []
+
+        # Extract time segments from coordinator data
+        for i in range(1, 10):  # Segments 1-9
+            segment = self._parse_time_segment(i)
+            time_segments.append(segment)
+
+        return time_segments
+
+    def _parse_time_segment(self, segment_id: int) -> dict:
+        """Parse a single time segment from coordinator data."""
+        # Get raw time values - these should always be present from the API
+        start_time_raw = self.data.get(f"forcedTimeStart{segment_id}")
+        end_time_raw = self.data.get(f"forcedTimeStop{segment_id}")
+
+        # Handle 'null' or empty values from API
+        if start_time_raw in ("null", None, ""):
+            start_time_raw = "0:0"
+        if end_time_raw in ("null", None, ""):
+            end_time_raw = "0:0"
+
+        # Format times with leading zeros (HH:MM)
+        start_time = self._format_time(str(start_time_raw))
+        end_time = self._format_time(str(end_time_raw))
+
+        # Get battery mode
+        batt_mode_int = int(
+            self.data.get(f"time{segment_id}Mode", BATT_MODE_LOAD_FIRST)
+        )
+
+        # Map numeric mode to string key (matches update_time_segment input format)
+        mode_map = {
+            BATT_MODE_LOAD_FIRST: "load_first",
+            BATT_MODE_BATTERY_FIRST: "battery_first",
+            BATT_MODE_GRID_FIRST: "grid_first",
+        }
+        batt_mode = mode_map.get(batt_mode_int, "load_first")
+
+        # Get enabled status
+        enabled = bool(int(self.data.get(f"forcedStopSwitch{segment_id}", 0)))
+
+        return {
+            "segment_id": segment_id,
+            "start_time": start_time,
+            "end_time": end_time,
+            "batt_mode": batt_mode,
+            "enabled": enabled,
+        }
+
+    def _format_time(self, time_raw: str) -> str:
+        """Format time string to HH:MM format."""
+        try:
+            parts = str(time_raw).split(":")
+            hour = int(parts[0])
+            minute = int(parts[1])
+        except (ValueError, IndexError):
+            return "00:00"
+        else:
+            return f"{hour:02d}:{minute:02d}"

homeassistant/components/growatt_server/icons.json

@@ -0,0 +1,10 @@
+{
+  "services": {
+    "read_time_segments": {
+      "service": "mdi:clock-outline"
+    },
+    "update_time_segment": {
+      "service": "mdi:clock-edit"
+    }
+  }
+}

homeassistant/components/growatt_server/services.py

@@ -0,0 +1,169 @@
+"""Service handlers for Growatt Server integration."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Any
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant, ServiceCall, SupportsResponse
+from homeassistant.exceptions import ServiceValidationError
+from homeassistant.helpers import device_registry as dr
+
+from .const import (
+    BATT_MODE_BATTERY_FIRST,
+    BATT_MODE_GRID_FIRST,
+    BATT_MODE_LOAD_FIRST,
+    DOMAIN,
+)
+
+if TYPE_CHECKING:
+    from .coordinator import GrowattCoordinator
+
+
+async def async_register_services(hass: HomeAssistant) -> None:
+    """Register services for Growatt Server integration."""
+
+    def get_min_coordinators() -> dict[str, GrowattCoordinator]:
+        """Get all MIN coordinators with V1 API from loaded config entries."""
+        min_coordinators: dict[str, GrowattCoordinator] = {}
+
+        for entry in hass.config_entries.async_entries(DOMAIN):
+            if entry.state != ConfigEntryState.LOADED:
+                continue
+
+            # Add MIN coordinators from this entry
+            for coord in entry.runtime_data.devices.values():
+                if coord.device_type == "min" and coord.api_version == "v1":
+                    min_coordinators[coord.device_id] = coord
+
+        return min_coordinators
+
+    def get_coordinator(device_id: str) -> GrowattCoordinator:
+        """Get coordinator by device_id.
+
+        Args:
+            device_id: Device registry ID (not serial number)
+        """
+        # Get current coordinators (they may have changed since service registration)
+        min_coordinators = get_min_coordinators()
+
+        if not min_coordinators:
+            raise ServiceValidationError(
+                "No MIN devices with token authentication are configured. "
+                "Services require MIN devices with V1 API access."
+            )
+
+        # Device registry ID provided - map to serial number
+        device_registry = dr.async_get(hass)
+        device_entry = device_registry.async_get(device_id)
+
+        if not device_entry:
+            raise ServiceValidationError(f"Device '{device_id}' not found")
+
+        # Extract serial number from device identifiers
+        serial_number = None
+        for identifier in device_entry.identifiers:
+            if identifier[0] == DOMAIN:
+                serial_number = identifier[1]
+                break
+
+        if not serial_number:
+            raise ServiceValidationError(
+                f"Device '{device_id}' is not a Growatt device"
+            )
+
+        # Find coordinator by serial number
+        if serial_number not in min_coordinators:
+            raise ServiceValidationError(
+                f"MIN device '{serial_number}' not found or not configured for services"
+            )
+
+        return min_coordinators[serial_number]
+
+    async def handle_update_time_segment(call: ServiceCall) -> None:
+        """Handle update_time_segment service call."""
+        segment_id: int = int(call.data["segment_id"])
+        batt_mode_str: str = call.data["batt_mode"]
+        start_time_str: str = call.data["start_time"]
+        end_time_str: str = call.data["end_time"]
+        enabled: bool = call.data["enabled"]
+        device_id: str = call.data["device_id"]
+
+        # Validate segment_id range
+        if not 1 <= segment_id <= 9:
+            raise ServiceValidationError(
+                f"segment_id must be between 1 and 9, got {segment_id}"
+            )
+
+        # Validate and convert batt_mode string to integer
+        valid_modes = {
+            "load_first": BATT_MODE_LOAD_FIRST,
+            "battery_first": BATT_MODE_BATTERY_FIRST,
+            "grid_first": BATT_MODE_GRID_FIRST,
+        }
+        if batt_mode_str not in valid_modes:
+            raise ServiceValidationError(
+                f"batt_mode must be one of {list(valid_modes.keys())}, got '{batt_mode_str}'"
+            )
+        batt_mode: int = valid_modes[batt_mode_str]
+
+        # Convert time strings to datetime.time objects
+        # UI time selector sends HH:MM:SS, but we only need HH:MM (strip seconds)
+        try:
+            # Take only HH:MM part (ignore seconds if present)
+            start_parts = start_time_str.split(":")
+            start_time_hhmm = f"{start_parts[0]}:{start_parts[1]}"
+            start_time = datetime.strptime(start_time_hhmm, "%H:%M").time()
+        except (ValueError, IndexError) as err:
+            raise ServiceValidationError(
+                "start_time must be in HH:MM or HH:MM:SS format"
+            ) from err
+
+        try:
+            # Take only HH:MM part (ignore seconds if present)
+            end_parts = end_time_str.split(":")
+            end_time_hhmm = f"{end_parts[0]}:{end_parts[1]}"
+            end_time = datetime.strptime(end_time_hhmm, "%H:%M").time()
+        except (ValueError, IndexError) as err:
+            raise ServiceValidationError(
+                "end_time must be in HH:MM or HH:MM:SS format"
+            ) from err
+
+        # Get the appropriate MIN coordinator
+        coordinator: GrowattCoordinator = get_coordinator(device_id)
+
+        await coordinator.update_time_segment(
+            segment_id,
+            batt_mode,
+            start_time,
+            end_time,
+            enabled,
+        )
+
+    async def handle_read_time_segments(call: ServiceCall) -> dict[str, Any]:
+        """Handle read_time_segments service call."""
+        device_id: str = call.data["device_id"]
+
+        # Get the appropriate MIN coordinator
+        coordinator: GrowattCoordinator = get_coordinator(device_id)
+
+        time_segments: list[dict[str, Any]] = await coordinator.read_time_segments()
+
+        return {"time_segments": time_segments}
+
+    # Register services without schema - services.yaml will provide UI definition
+    # Schema validation happens in the handler functions
+    hass.services.async_register(
+        DOMAIN,
+        "update_time_segment",
+        handle_update_time_segment,
+        supports_response=SupportsResponse.NONE,
+    )
+
+    hass.services.async_register(
+        DOMAIN,
+        "read_time_segments",
+        handle_read_time_segments,
+        supports_response=SupportsResponse.ONLY,
+    )