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

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

homeassistant/components/growatt_server/__init__.py

@@ -10,6 +10,8 @@ from requests import RequestException
 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 @@ from .const import (
     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,8 @@ ERROR_INVALID_AUTH = "invalid_auth"
 
 # 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

homeassistant/components/growatt_server/coordinator.py

@@ -12,10 +12,17 @@ import growattServer
 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 @@ class GrowattCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         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,
+    )

homeassistant/components/growatt_server/services.yaml

@@ -0,0 +1,50 @@
+# Service definitions for Growatt Server integration
+
+update_time_segment:
+  fields:
+    segment_id:
+      required: true
+      example: 1
+      selector:
+        number:
+          min: 1
+          max: 9
+          mode: box
+    batt_mode:
+      required: true
+      example: "load_first"
+      selector:
+        select:
+          options:
+            - "load_first"
+            - "battery_first"
+            - "grid_first"
+          translation_key: batt_mode
+    start_time:
+      required: true
+      example: "08:00"
+      selector:
+        time:
+    end_time:
+      required: true
+      example: "12:00"
+      selector:
+        time:
+    enabled:
+      required: true
+      example: true
+      selector:
+        boolean:
+    device_id:
+      required: true
+      selector:
+        device:
+          integration: growatt_server
+
+read_time_segments:
+  fields:
+    device_id:
+      required: true
+      selector:
+        device:
+          integration: growatt_server

homeassistant/components/growatt_server/strings.json

@@ -523,5 +523,56 @@
       }
     }
   },
+  "selector": {
+    "batt_mode": {
+      "options": {
+        "battery_first": "Battery first",
+        "grid_first": "Grid first",
+        "load_first": "Load first"
+      }
+    }
+  },
+  "services": {
+    "read_time_segments": {
+      "description": "Read all time segments from a supported inverter.",
+      "fields": {
+        "device_id": {
+          "description": "The Growatt device to perform the action on.",
+          "name": "Device"
+        }
+      },
+      "name": "Read time segments"
+    },
+    "update_time_segment": {
+      "description": "Update a time segment for supported inverters.",
+      "fields": {
+        "batt_mode": {
+          "description": "Battery operation mode for this time segment.",
+          "name": "Battery mode"
+        },
+        "device_id": {
+          "description": "[%key:component::growatt_server::services::read_time_segments::fields::device_id::description%]",
+          "name": "[%key:component::growatt_server::services::read_time_segments::fields::device_id::name%]"
+        },
+        "enabled": {
+          "description": "Whether this time segment is active.",
+          "name": "Enabled"
+        },
+        "end_time": {
+          "description": "End time for the segment (HH:MM format).",
+          "name": "End time"
+        },
+        "segment_id": {
+          "description": "Time segment ID (1-9).",
+          "name": "Segment ID"
+        },
+        "start_time": {
+          "description": "Start time for the segment (HH:MM format).",
+          "name": "Start time"
+        }
+      },
+      "name": "Update time segment"
+    }
+  },
   "title": "Growatt Server"
 }

tests/components/growatt_server/conftest.py

@@ -1,6 +1,6 @@
 """Common fixtures for the Growatt server tests."""
 
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -36,6 +36,9 @@ def mock_growatt_v1_api():
 
     Methods mocked for switch and number operations:
     - min_write_parameter: Called by switch/number entities to change settings
+
+    Methods mocked for service operations:
+    - min_write_time_segment: Called by time segment management services
     """
     with patch(
         "homeassistant.components.growatt_server.config_flow.growattServer.OpenApiV1",
@@ -65,15 +68,45 @@ def mock_growatt_v1_api():
 
         # Called by MIN device coordinator during refresh
         mock_v1_api.min_settings.return_value = {
-            # Forced charge time segments (not used by switch/number, but coordinator fetches it)
+            # Time segment 1 - enabled, load_first mode
             "forcedTimeStart1": "06:00",
             "forcedTimeStop1": "08:00",
-            "forcedChargeBatMode1": 1,
-            "forcedChargeFlag1": 1,
+            "time1Mode": 1,  # load_first
+            "forcedStopSwitch1": 1,  # enabled
+            # Time segment 2 - disabled
             "forcedTimeStart2": "22:00",
             "forcedTimeStop2": "24:00",
-            "forcedChargeBatMode2": 0,
-            "forcedChargeFlag2": 0,
+            "time2Mode": 0,  # battery_first
+            "forcedStopSwitch2": 0,  # disabled
+            # Time segments 3-9 - all disabled with default values
+            "forcedTimeStart3": "00:00",
+            "forcedTimeStop3": "00:00",
+            "time3Mode": 1,
+            "forcedStopSwitch3": 0,
+            "forcedTimeStart4": "00:00",
+            "forcedTimeStop4": "00:00",
+            "time4Mode": 1,
+            "forcedStopSwitch4": 0,
+            "forcedTimeStart5": "00:00",
+            "forcedTimeStop5": "00:00",
+            "time5Mode": 1,
+            "forcedStopSwitch5": 0,
+            "forcedTimeStart6": "00:00",
+            "forcedTimeStop6": "00:00",
+            "time6Mode": 1,
+            "forcedStopSwitch6": 0,
+            "forcedTimeStart7": "00:00",
+            "forcedTimeStop7": "00:00",
+            "time7Mode": 1,
+            "forcedStopSwitch7": 0,
+            "forcedTimeStart8": "00:00",
+            "forcedTimeStop8": "00:00",
+            "time8Mode": 1,
+            "forcedStopSwitch8": 0,
+            "forcedTimeStart9": "00:00",
+            "forcedTimeStop9": "00:00",
+            "time9Mode": 1,
+            "forcedStopSwitch9": 0,
         }
 
         # Called by MIN device coordinator during refresh
@@ -101,6 +134,15 @@ def mock_growatt_v1_api():
         # Called by switch/number entities during turn_on/turn_off/set_value
         mock_v1_api.min_write_parameter.return_value = None
 
+        # Called by time segment management services
+        # Note: Don't use autospec for this method as it needs to accept variable arguments
+        mock_v1_api.min_write_time_segment = Mock(
+            return_value={
+                "error_code": 0,
+                "error_msg": "Success",
+            }
+        )
+
         yield mock_v1_api
 
 

tests/components/growatt_server/snapshots/test_services.ambr

@@ -0,0 +1,70 @@
+# serializer version: 1
+# name: test_read_time_segments_single_device
+  dict({
+    'time_segments': list([
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': True,
+        'end_time': '08:00',
+        'segment_id': 1,
+        'start_time': '06:00',
+      }),
+      dict({
+        'batt_mode': 'load_first',
+        'enabled': False,
+        'end_time': '24:00',
+        'segment_id': 2,
+        'start_time': '22:00',
+      }),
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': False,
+        'end_time': '00:00',
+        'segment_id': 3,
+        'start_time': '00:00',
+      }),
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': False,
+        'end_time': '00:00',
+        'segment_id': 4,
+        'start_time': '00:00',
+      }),
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': False,
+        'end_time': '00:00',
+        'segment_id': 5,
+        'start_time': '00:00',
+      }),
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': False,
+        'end_time': '00:00',
+        'segment_id': 6,
+        'start_time': '00:00',
+      }),
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': False,
+        'end_time': '00:00',
+        'segment_id': 7,
+        'start_time': '00:00',
+      }),
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': False,
+        'end_time': '00:00',
+        'segment_id': 8,
+        'start_time': '00:00',
+      }),
+      dict({
+        'batt_mode': 'battery_first',
+        'enabled': False,
+        'end_time': '00:00',
+        'segment_id': 9,
+        'start_time': '00:00',
+      }),
+    ]),
+  })
+# ---

tests/components/growatt_server/test_services.py

@@ -0,0 +1,586 @@
+"""Test Growatt Server services."""
+
+from unittest.mock import patch
+
+import growattServer
+import pytest
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.growatt_server.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import device_registry as dr
+
+from tests.common import MockConfigEntry
+
+
+@pytest.mark.usefixtures("mock_growatt_v1_api")
+async def test_read_time_segments_single_device(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test reading time segments for single device."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Test service call
+    response = await hass.services.async_call(
+        DOMAIN,
+        "read_time_segments",
+        {"device_id": device_entry.id},
+        blocking=True,
+        return_response=True,
+    )
+
+    assert response == snapshot
+
+
+async def test_update_time_segment_charge_mode(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_growatt_v1_api,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test updating time segment with charge mode."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Test successful update
+    await hass.services.async_call(
+        DOMAIN,
+        "update_time_segment",
+        {
+            "device_id": device_entry.id,
+            "segment_id": 1,
+            "start_time": "09:00",
+            "end_time": "11:00",
+            "batt_mode": "load_first",
+            "enabled": True,
+        },
+        blocking=True,
+    )
+
+    # Verify the API was called
+    mock_growatt_v1_api.min_write_time_segment.assert_called_once()
+
+
+async def test_update_time_segment_discharge_mode(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_growatt_v1_api,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test updating time segment with discharge mode."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    await hass.services.async_call(
+        DOMAIN,
+        "update_time_segment",
+        {
+            "device_id": device_entry.id,
+            "segment_id": 2,
+            "start_time": "14:00",
+            "end_time": "16:00",
+            "batt_mode": "battery_first",
+            "enabled": True,
+        },
+        blocking=True,
+    )
+
+    mock_growatt_v1_api.min_write_time_segment.assert_called_once()
+
+
+async def test_update_time_segment_standby_mode(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_growatt_v1_api,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test updating time segment with standby mode."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    await hass.services.async_call(
+        DOMAIN,
+        "update_time_segment",
+        {
+            "device_id": device_entry.id,
+            "segment_id": 3,
+            "start_time": "20:00",
+            "end_time": "22:00",
+            "batt_mode": "grid_first",
+            "enabled": True,
+        },
+        blocking=True,
+    )
+
+    mock_growatt_v1_api.min_write_time_segment.assert_called_once()
+
+
+async def test_update_time_segment_disabled(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_growatt_v1_api,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test disabling a time segment."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    await hass.services.async_call(
+        DOMAIN,
+        "update_time_segment",
+        {
+            "device_id": device_entry.id,
+            "segment_id": 1,
+            "start_time": "06:00",
+            "end_time": "08:00",
+            "batt_mode": "load_first",
+            "enabled": False,
+        },
+        blocking=True,
+    )
+
+    mock_growatt_v1_api.min_write_time_segment.assert_called_once()
+
+
+async def test_update_time_segment_with_seconds(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_growatt_v1_api,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test updating time segment with HH:MM:SS format from UI."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Test with HH:MM:SS format (what the UI time selector sends)
+    await hass.services.async_call(
+        DOMAIN,
+        "update_time_segment",
+        {
+            "device_id": device_entry.id,
+            "segment_id": 1,
+            "start_time": "09:00:00",
+            "end_time": "11:00:00",
+            "batt_mode": "load_first",
+            "enabled": True,
+        },
+        blocking=True,
+    )
+
+    mock_growatt_v1_api.min_write_time_segment.assert_called_once()
+
+
+async def test_update_time_segment_api_error(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_growatt_v1_api,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test handling API error when updating time segment."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Mock API error - the library raises an exception instead of returning error dict
+    mock_growatt_v1_api.min_write_time_segment.side_effect = (
+        growattServer.GrowattV1ApiError(
+            "Error during writing time segment 1",
+            error_code=1,
+            error_msg="API Error",
+        )
+    )
+
+    with pytest.raises(HomeAssistantError, match="API error updating time segment"):
+        await hass.services.async_call(
+            DOMAIN,
+            "update_time_segment",
+            {
+                "device_id": device_entry.id,
+                "segment_id": 1,
+                "start_time": "09:00",
+                "end_time": "11:00",
+                "batt_mode": "load_first",
+                "enabled": True,
+            },
+            blocking=True,
+        )
+
+
+@pytest.mark.usefixtures("mock_growatt_classic_api")
+async def test_no_min_devices_skips_service_registration(
+    hass: HomeAssistant,
+    mock_config_entry_classic: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test that services fail gracefully when no MIN devices exist."""
+    mock_config_entry_classic.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        # Only non-MIN devices (TLX with classic API)
+        mock_get_devices.return_value = (
+            [{"deviceSn": "TLX123456", "deviceType": "tlx"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry_classic.entry_id)
+        await hass.async_block_till_done()
+
+    # Verify services are registered (they're always registered in async_setup)
+    assert hass.services.has_service(DOMAIN, "update_time_segment")
+    assert hass.services.has_service(DOMAIN, "read_time_segments")
+
+    # Get the TLX device (non-MIN)
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "TLX123456")})
+    assert device_entry is not None
+
+    # But calling them with a non-MIN device should fail with appropriate error
+    with pytest.raises(
+        ServiceValidationError, match="No MIN devices with token authentication"
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "update_time_segment",
+            {
+                "device_id": device_entry.id,
+                "segment_id": 1,
+                "start_time": "09:00",
+                "end_time": "11:00",
+                "batt_mode": "load_first",
+                "enabled": True,
+            },
+            blocking=True,
+        )
+
+
+async def test_multiple_devices_with_valid_device_id_works(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_growatt_v1_api,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test that multiple devices work when device_id is specified."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [
+                {"deviceSn": "MIN123456", "deviceType": "min"},
+                {"deviceSn": "MIN789012", "deviceType": "min"},
+            ],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID for the first MIN device
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Test update service with specific device_id (device registry ID)
+    await hass.services.async_call(
+        DOMAIN,
+        "update_time_segment",
+        {
+            "device_id": device_entry.id,
+            "segment_id": 1,
+            "start_time": "09:00",
+            "end_time": "11:00",
+            "batt_mode": "load_first",
+            "enabled": True,
+        },
+        blocking=True,
+    )
+
+    mock_growatt_v1_api.min_write_time_segment.assert_called_once()
+
+    # Test read service with specific device_id (device registry ID)
+    response = await hass.services.async_call(
+        DOMAIN,
+        "read_time_segments",
+        {"device_id": device_entry.id},
+        blocking=True,
+        return_response=True,
+    )
+
+    assert response is not None
+    assert "time_segments" in response
+
+
+@pytest.mark.usefixtures("mock_growatt_v1_api")
+async def test_update_time_segment_invalid_time_format(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test handling invalid time format in update_time_segment."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Test with invalid time format
+    with pytest.raises(
+        ServiceValidationError, match="start_time must be in HH:MM or HH:MM:SS format"
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "update_time_segment",
+            {
+                "device_id": device_entry.id,
+                "segment_id": 1,
+                "start_time": "invalid",
+                "end_time": "11:00",
+                "batt_mode": "load_first",
+                "enabled": True,
+            },
+            blocking=True,
+        )
+
+
+@pytest.mark.usefixtures("mock_growatt_v1_api")
+async def test_update_time_segment_invalid_segment_id(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test validation of segment_id range."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Test segment_id too low
+    with pytest.raises(
+        ServiceValidationError, match="segment_id must be between 1 and 9"
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "update_time_segment",
+            {
+                "device_id": device_entry.id,
+                "segment_id": 0,
+                "start_time": "09:00",
+                "end_time": "11:00",
+                "batt_mode": "load_first",
+                "enabled": True,
+            },
+            blocking=True,
+        )
+
+    # Test segment_id too high
+    with pytest.raises(
+        ServiceValidationError, match="segment_id must be between 1 and 9"
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "update_time_segment",
+            {
+                "device_id": device_entry.id,
+                "segment_id": 10,
+                "start_time": "09:00",
+                "end_time": "11:00",
+                "batt_mode": "load_first",
+                "enabled": True,
+            },
+            blocking=True,
+        )
+
+
+@pytest.mark.usefixtures("mock_growatt_v1_api")
+async def test_update_time_segment_invalid_batt_mode(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test validation of batt_mode value."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Test invalid batt_mode
+    with pytest.raises(ServiceValidationError, match="batt_mode must be one of"):
+        await hass.services.async_call(
+            DOMAIN,
+            "update_time_segment",
+            {
+                "device_id": device_entry.id,
+                "segment_id": 1,
+                "start_time": "09:00",
+                "end_time": "11:00",
+                "batt_mode": "invalid_mode",
+                "enabled": True,
+            },
+            blocking=True,
+        )
+
+
+@pytest.mark.usefixtures("mock_growatt_v1_api")
+async def test_read_time_segments_api_error(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test handling API error when reading time segments."""
+    mock_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.growatt_server.get_device_list"
+    ) as mock_get_devices:
+        mock_get_devices.return_value = (
+            [{"deviceSn": "MIN123456", "deviceType": "min"}],
+            "12345",
+        )
+        assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Get the device registry ID
+    device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "MIN123456")})
+    assert device_entry is not None
+
+    # Mock API error by making coordinator.read_time_segments raise an exception
+    with (
+        patch(
+            "homeassistant.components.growatt_server.coordinator.GrowattCoordinator.read_time_segments",
+            side_effect=HomeAssistantError("API connection failed"),
+        ),
+        pytest.raises(HomeAssistantError, match="API connection failed"),
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            "read_time_segments",
+            {"device_id": device_entry.id},
+            blocking=True,
+            return_response=True,
+        )