Add feeder meal plan actions to tuya (#161488)

Co-authored-by: Norbert Rittel <norbert@rittel.de>
Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

homeassistant/components/tuya/__init__.py

@@ -17,8 +17,9 @@ from tuya_sharing import (
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import ConfigEntryAuthFailed
-from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers import config_validation as cv, device_registry as dr
 from homeassistant.helpers.dispatcher import dispatcher_send
+from homeassistant.helpers.typing import ConfigType
 
 from .const import (
     CONF_ENDPOINT,
@@ -32,6 +33,9 @@ from .const import (
     TUYA_DISCOVERY_NEW,
     TUYA_HA_SIGNAL_UPDATE_ENTITY,
 )
+from .services import async_setup_services
+
+CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 
 # Suppress logs from the library, it logs unneeded on error
 logging.getLogger("tuya_sharing").setLevel(logging.CRITICAL)
@@ -58,6 +62,13 @@ def _create_manager(entry: TuyaConfigEntry, token_listener: TokenListener) -> Ma
     )
 
 
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the Tuya Services."""
+    await async_setup_services(hass)
+
+    return True
+
+
 async def async_setup_entry(hass: HomeAssistant, entry: TuyaConfigEntry) -> bool:
     """Async setup hass config entry."""
     await hass.async_add_executor_job(

homeassistant/components/tuya/icons.json

@@ -381,5 +381,13 @@
         "default": "mdi:watermark"
       }
     }
+  },
+  "services": {
+    "get_feeder_meal_plan": {
+      "service": "mdi:database-eye"
+    },
+    "set_feeder_meal_plan": {
+      "service": "mdi:database-edit"
+    }
   }
 }

homeassistant/components/tuya/services.py

@@ -0,0 +1,160 @@
+"""Services for Tuya integration."""
+
+from enum import StrEnum
+from typing import Any
+
+from tuya_device_handlers.device_wrapper.service_feeder_schedule import (
+    FeederSchedule,
+    get_feeder_schedule_wrapper,
+)
+from tuya_sharing import CustomerDevice, Manager
+import voluptuous as vol
+
+from homeassistant.const import ATTR_DEVICE_ID
+from homeassistant.core import HomeAssistant, ServiceCall, SupportsResponse
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import device_registry as dr
+
+from .const import DOMAIN
+
+DAYS = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
+
+FEEDING_ENTRY_SCHEMA = vol.Schema(
+    {
+        vol.Optional("days"): [vol.In(DAYS)],
+        vol.Required("time"): str,
+        vol.Required("portion"): int,
+        vol.Required("enabled"): bool,
+    }
+)
+
+
+class Service(StrEnum):
+    """Tuya services."""
+
+    GET_FEEDER_MEAL_PLAN = "get_feeder_meal_plan"
+    SET_FEEDER_MEAL_PLAN = "set_feeder_meal_plan"
+
+
+def _get_tuya_device(
+    hass: HomeAssistant, device_id: str
+) -> tuple[CustomerDevice, Manager]:
+    """Get a Tuya device and manager from a Home Assistant device registry ID."""
+    device_registry = dr.async_get(hass)
+    device_entry = device_registry.async_get(device_id)
+    if device_entry is None:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="device_not_found",
+            translation_placeholders={
+                "device_id": device_id,
+            },
+        )
+
+    # Find the Tuya device ID from identifiers
+    tuya_device_id = None
+    for identifier_domain, identifier_value in device_entry.identifiers:
+        if identifier_domain == DOMAIN:
+            tuya_device_id = identifier_value
+            break
+
+    if tuya_device_id is None:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="device_not_tuya_device",
+            translation_placeholders={
+                "device_id": device_id,
+            },
+        )
+
+    # Find the device in Tuya config entry
+    for entry in hass.config_entries.async_loaded_entries(DOMAIN):
+        manager = entry.runtime_data.manager
+        if tuya_device_id in manager.device_map:
+            return manager.device_map[tuya_device_id], manager
+
+    raise ServiceValidationError(
+        translation_domain=DOMAIN,
+        translation_key="device_not_found",
+        translation_placeholders={
+            "device_id": device_id,
+        },
+    )
+
+
+async def async_get_feeder_meal_plan(
+    call: ServiceCall,
+) -> dict[str, Any]:
+    """Handle get_feeder_meal_plan service call."""
+    device, _ = _get_tuya_device(call.hass, call.data[ATTR_DEVICE_ID])
+
+    if not (wrapper := get_feeder_schedule_wrapper(device)):
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="device_not_support_meal_plan_status",
+            translation_placeholders={
+                "device_id": device.id,
+            },
+        )
+
+    meal_plan = wrapper.read_device_status(device)
+    if meal_plan is None:
+        raise HomeAssistantError(
+            translation_domain=DOMAIN,
+            translation_key="invalid_meal_plan_data",
+        )
+
+    return {"meal_plan": meal_plan}
+
+
+async def async_set_feeder_meal_plan(call: ServiceCall) -> None:
+    """Handle set_feeder_meal_plan service call."""
+    device, manager = _get_tuya_device(call.hass, call.data[ATTR_DEVICE_ID])
+
+    if not (wrapper := get_feeder_schedule_wrapper(device)):
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="device_not_support_meal_plan_function",
+            translation_placeholders={
+                "device_id": device.id,
+            },
+        )
+
+    meal_plan: list[FeederSchedule] = call.data["meal_plan"]
+
+    await call.hass.async_add_executor_job(
+        manager.send_commands,
+        device.id,
+        wrapper.get_update_commands(device, meal_plan),
+    )
+
+
+async def async_setup_services(hass: HomeAssistant) -> None:
+    """Set up Tuya services."""
+
+    hass.services.async_register(
+        DOMAIN,
+        Service.GET_FEEDER_MEAL_PLAN,
+        async_get_feeder_meal_plan,
+        schema=vol.Schema(
+            {
+                vol.Required(ATTR_DEVICE_ID): str,
+            }
+        ),
+        supports_response=SupportsResponse.ONLY,
+    )
+
+    hass.services.async_register(
+        DOMAIN,
+        Service.SET_FEEDER_MEAL_PLAN,
+        async_set_feeder_meal_plan,
+        schema=vol.Schema(
+            {
+                vol.Required(ATTR_DEVICE_ID): str,
+                vol.Required("meal_plan"): vol.All(
+                    list,
+                    [FEEDING_ENTRY_SCHEMA],
+                ),
+            }
+        ),
+    )

homeassistant/components/tuya/services.yaml

@@ -0,0 +1,51 @@
+get_feeder_meal_plan:
+  fields:
+    device_id:
+      required: true
+      selector:
+        device:
+          integration: tuya
+
+set_feeder_meal_plan:
+  fields:
+    device_id:
+      required: true
+      selector:
+        device:
+          integration: tuya
+    meal_plan:
+      required: true
+      selector:
+        object:
+          translation_key: set_feeder_meal_plan
+          description_field: portion
+          multiple: true
+          fields:
+            days:
+              selector:
+                select:
+                  options:
+                    - monday
+                    - tuesday
+                    - wednesday
+                    - thursday
+                    - friday
+                    - saturday
+                    - sunday
+                  multiple: true
+                  translation_key: days_of_week
+
+            time:
+              selector:
+                time:
+
+            portion:
+              selector:
+                number:
+                  min: 0
+                  max: 100
+                  mode: box
+                  unit_of_measurement: "g"
+            enabled:
+              selector:
+                boolean: {}

homeassistant/components/tuya/strings.json

@@ -1099,6 +1099,80 @@
   "exceptions": {
     "action_dpcode_not_found": {
       "message": "Unable to process action as the device does not provide a corresponding function code (expected one of {expected} in {available})."
+    },
+    "device_not_found": {
+      "message": "Feeder with ID {device_id} could not be found."
+    },
+    "device_not_support_meal_plan_function": {
+      "message": "Feeder with ID {device_id} does not support meal plan functionality."
+    },
+    "device_not_support_meal_plan_status": {
+      "message": "Feeder with ID {device_id} does not support meal plan status."
+    },
+    "device_not_tuya_device": {
+      "message": "Device with ID {device_id} is not a Tuya feeder."
+    },
+    "invalid_meal_plan_data": {
+      "message": "Unable to parse meal plan data."
+    }
+  },
+  "selector": {
+    "days_of_week": {
+      "options": {
+        "friday": "[%key:common::time::friday%]",
+        "monday": "[%key:common::time::monday%]",
+        "saturday": "[%key:common::time::saturday%]",
+        "sunday": "[%key:common::time::sunday%]",
+        "thursday": "[%key:common::time::thursday%]",
+        "tuesday": "[%key:common::time::tuesday%]",
+        "wednesday": "[%key:common::time::wednesday%]"
+      }
+    },
+    "set_feeder_meal_plan": {
+      "fields": {
+        "days": {
+          "description": "Days of the week for the meal plan.",
+          "name": "Days"
+        },
+        "enabled": {
+          "description": "Whether the meal plan is enabled.",
+          "name": "Enabled"
+        },
+        "portion": {
+          "description": "Amount in grams",
+          "name": "Portion"
+        },
+        "time": {
+          "description": "Time of the meal.",
+          "name": "Time"
+        }
+      }
+    }
+  },
+  "services": {
+    "get_feeder_meal_plan": {
+      "description": "Retrieves a meal plan from a Tuya feeder.",
+      "fields": {
+        "device_id": {
+          "description": "The Tuya feeder.",
+          "name": "[%key:common::config_flow::data::device%]"
+        }
+      },
+      "name": "Get feeder meal plan data"
+    },
+    "set_feeder_meal_plan": {
+      "description": "Sets a meal plan on a Tuya feeder.",
+      "fields": {
+        "device_id": {
+          "description": "[%key:component::tuya::services::get_feeder_meal_plan::fields::device_id::description%]",
+          "name": "[%key:common::config_flow::data::device%]"
+        },
+        "meal_plan": {
+          "description": "The meal plan data to set.",
+          "name": "Meal plan"
+        }
+      },
+      "name": "Set feeder meal plan data"
     }
   }
 }

tests/components/tuya/snapshots/test_services.ambr

@@ -0,0 +1,113 @@
+# serializer version: 1
+# name: test_get_feeder_meal_plan[cwwsq_wfkzyy0evslzsmoi]
+  dict({
+    'meal_plan': list([
+      dict({
+        'days': list([
+          'monday',
+          'tuesday',
+          'wednesday',
+          'thursday',
+          'friday',
+          'saturday',
+          'sunday',
+        ]),
+        'enabled': False,
+        'portion': 2,
+        'time': '04:00',
+      }),
+      dict({
+        'days': list([
+          'monday',
+          'tuesday',
+          'wednesday',
+          'thursday',
+          'friday',
+          'saturday',
+          'sunday',
+        ]),
+        'enabled': False,
+        'portion': 1,
+        'time': '06:00',
+      }),
+      dict({
+        'days': list([
+          'monday',
+          'tuesday',
+          'wednesday',
+          'thursday',
+          'friday',
+          'saturday',
+          'sunday',
+        ]),
+        'enabled': True,
+        'portion': 2,
+        'time': '09:00',
+      }),
+      dict({
+        'days': list([
+          'monday',
+          'tuesday',
+          'wednesday',
+          'thursday',
+          'friday',
+          'saturday',
+          'sunday',
+        ]),
+        'enabled': False,
+        'portion': 1,
+        'time': '12:00',
+      }),
+      dict({
+        'days': list([
+          'monday',
+          'tuesday',
+          'wednesday',
+          'thursday',
+          'friday',
+          'saturday',
+          'sunday',
+        ]),
+        'enabled': True,
+        'portion': 2,
+        'time': '15:00',
+      }),
+      dict({
+        'days': list([
+          'monday',
+          'tuesday',
+          'wednesday',
+          'thursday',
+          'friday',
+          'saturday',
+          'sunday',
+        ]),
+        'enabled': True,
+        'portion': 2,
+        'time': '21:00',
+      }),
+      dict({
+        'days': list([
+          'monday',
+          'tuesday',
+          'wednesday',
+          'thursday',
+          'friday',
+          'saturday',
+          'sunday',
+        ]),
+        'enabled': False,
+        'portion': 1,
+        'time': '23:00',
+      }),
+      dict({
+        'days': list([
+          'thursday',
+        ]),
+        'enabled': True,
+        'portion': 1,
+        'time': '18:00',
+      }),
+    ]),
+  })
+# ---

tests/components/tuya/test_services.py

@@ -0,0 +1,250 @@
+"""Tests for Tuya services."""
+
+from __future__ import annotations
+
+import pytest
+from syrupy.assertion import SnapshotAssertion
+from tuya_device_handlers.device_wrapper.service_feeder_schedule import FeederSchedule
+from tuya_sharing import CustomerDevice, Manager
+
+from homeassistant.components.tuya.const import DOMAIN
+from homeassistant.components.tuya.services import Service
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import device_registry as dr
+
+from . import initialize_entry
+
+from tests.common import MockConfigEntry
+
+DECODED_MEAL_PLAN: list[FeederSchedule] = [
+    {
+        "days": [
+            "monday",
+            "tuesday",
+            "wednesday",
+            "thursday",
+            "friday",
+            "saturday",
+            "sunday",
+        ],
+        "time": "09:00",
+        "portion": 1,
+        "enabled": True,
+    },
+    {
+        "days": [
+            "monday",
+            "tuesday",
+            "wednesday",
+            "thursday",
+            "friday",
+            "saturday",
+            "sunday",
+        ],
+        "time": "09:30",
+        "portion": 1,
+        "enabled": True,
+    },
+]
+
+
+@pytest.mark.parametrize("mock_device_code", ["cwwsq_wfkzyy0evslzsmoi"])
+async def test_get_feeder_meal_plan(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+    snapshot: SnapshotAssertion,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test GET_FEEDER_MEAL_PLAN with valid meal plan data."""
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+    device_entry = device_registry.async_get_device(
+        identifiers={(DOMAIN, mock_device.id)}
+    )
+    assert device_entry is not None
+    device_id = device_entry.id
+
+    result = await hass.services.async_call(
+        DOMAIN,
+        Service.GET_FEEDER_MEAL_PLAN,
+        {"device_id": device_id},
+        blocking=True,
+        return_response=True,
+    )
+    assert result == snapshot
+
+
+@pytest.mark.parametrize("mock_device_code", ["cwwsq_wfkzyy0evslzsmoi"])
+async def test_get_feeder_meal_plan_invalid_meal_plan(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test GET_FEEDER_MEAL_PLAN error when meal plan data is missing."""
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+    device_entry = device_registry.async_get_device(
+        identifiers={(DOMAIN, mock_device.id)}
+    )
+    assert device_entry is not None
+    device_id = device_entry.id
+
+    mock_device.status.pop("meal_plan", None)
+    with pytest.raises(
+        HomeAssistantError,
+        match="Unable to parse meal plan data",
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            Service.GET_FEEDER_MEAL_PLAN,
+            {"device_id": device_id},
+            blocking=True,
+            return_response=True,
+        )
+
+
+@pytest.mark.parametrize("mock_device_code", ["cwwsq_wfkzyy0evslzsmoi"])
+async def test_set_feeder_meal_plan(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test SET_FEEDER_MEAL_PLAN with valid device and meal plan data."""
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+    device_entry = device_registry.async_get_device(
+        identifiers={(DOMAIN, mock_device.id)}
+    )
+    assert device_entry is not None
+    device_id = device_entry.id
+
+    await hass.services.async_call(
+        DOMAIN,
+        Service.SET_FEEDER_MEAL_PLAN,
+        {
+            "device_id": device_id,
+            "meal_plan": DECODED_MEAL_PLAN,
+        },
+        blocking=True,
+    )
+    mock_manager.send_commands.assert_called_once_with(
+        mock_device.id,
+        [{"code": "meal_plan", "value": "fwkAAQF/CR4BAQ=="}],
+    )
+
+
+@pytest.mark.parametrize("mock_device_code", ["cwwsq_wfkzyy0evslzsmoi"])
+async def test_set_feeder_meal_plan_unsupported_device(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test SET_FEEDER_MEAL_PLAN error when device is unsupported."""
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+    device_entry = device_registry.async_get_device(
+        identifiers={(DOMAIN, mock_device.id)}
+    )
+    assert device_entry is not None
+    device_id = device_entry.id
+
+    mock_device.product_id = "unsupported_product"
+    with pytest.raises(
+        ServiceValidationError,
+        match=f"Feeder with ID {mock_device.id} does not support meal plan functionality",
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            Service.SET_FEEDER_MEAL_PLAN,
+            {
+                "device_id": device_id,
+                "meal_plan": DECODED_MEAL_PLAN,
+            },
+            blocking=True,
+        )
+
+
+@pytest.mark.parametrize("mock_device_code", ["cwwsq_wfkzyy0evslzsmoi"])
+async def test_get_tuya_device_error_device_not_found(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+) -> None:
+    """Test service error when device ID does not exist."""
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+
+    with pytest.raises(
+        ServiceValidationError,
+        match="Feeder with ID invalid_device_id could not be found",
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            Service.GET_FEEDER_MEAL_PLAN,
+            {"device_id": "invalid_device_id"},
+            blocking=True,
+            return_response=True,
+        )
+
+
+@pytest.mark.parametrize("mock_device_code", ["cwwsq_wfkzyy0evslzsmoi"])
+async def test_get_tuya_device_error_non_tuya_device(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+) -> None:
+    """Test service error when target device is not a Tuya device."""
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+
+    device_registry = dr.async_get(hass)
+    non_tuya_device = device_registry.async_get_or_create(
+        config_entry_id=mock_config_entry.entry_id,
+        identifiers={("other_domain", "some_id")},
+        name="Non-Tuya Device",
+    )
+    with pytest.raises(
+        ServiceValidationError,
+        match=f"Device with ID {non_tuya_device.id} is not a Tuya feeder",
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            Service.GET_FEEDER_MEAL_PLAN,
+            {"device_id": non_tuya_device.id},
+            blocking=True,
+            return_response=True,
+        )
+
+
+@pytest.mark.parametrize("mock_device_code", ["cwwsq_wfkzyy0evslzsmoi"])
+async def test_get_tuya_device_error_unknown_tuya_device(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+) -> None:
+    """Test service error when Tuya identifier is not present in manager map."""
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+
+    device_registry = dr.async_get(hass)
+    tuya_device = device_registry.async_get_or_create(
+        config_entry_id=mock_config_entry.entry_id,
+        identifiers={(DOMAIN, "unknown_tuya_id")},
+        name="Unknown Tuya Device",
+    )
+    with pytest.raises(
+        ServiceValidationError,
+        match=f"Feeder with ID {tuya_device.id} could not be found",
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            Service.GET_FEEDER_MEAL_PLAN,
+            {"device_id": tuya_device.id},
+            blocking=True,
+            return_response=True,
+        )