Add Energyzero get_prices service (#100499)

homeassistant/components/energyzero/__init__.py

@@ -1,17 +1,74 @@
 """The EnergyZero integration."""
 from __future__ import annotations
 
+from datetime import date, datetime
+
+from energyzero import Electricity, EnergyZero, Gas
+
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
-from homeassistant.core import HomeAssistant
+from homeassistant.core import (
+    HomeAssistant,
+    ServiceCall,
+    ServiceResponse,
+    SupportsResponse,
+)
 from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.util import dt as dt_util
 
-from .const import DOMAIN
+from .const import (
+    ATTR_END,
+    ATTR_INCL_VAT,
+    ATTR_START,
+    ATTR_TYPE,
+    DOMAIN,
+    SERVICE_NAME,
+    SERVICE_SCHEMA,
+)
 from .coordinator import EnergyZeroDataUpdateCoordinator
 
 PLATFORMS = [Platform.SENSOR]
 
 
+def _get_date(date_input: str | None) -> date | datetime:
+    """Get date."""
+    if not date_input:
+        return dt_util.now().date()
+
+    if value := dt_util.parse_datetime(date_input):
+        return value
+
+    raise ValueError(f"Invalid date: {date_input}")
+
+
+def __serialize_prices(prices: Electricity | Gas) -> ServiceResponse:
+    """Serialize prices."""
+    return {str(timestamp): price for timestamp, price in prices.prices.items()}
+
+
+async def _get_prices(hass: HomeAssistant, call: ServiceCall) -> ServiceResponse:
+    """Search prices."""
+    price_type = call.data[ATTR_TYPE]
+
+    energyzero = EnergyZero(
+        session=async_get_clientsession(hass),
+        incl_btw=str(call.data[ATTR_INCL_VAT]).lower(),
+    )
+
+    start = _get_date(call.data.get(ATTR_START))
+    end = _get_date(call.data.get(ATTR_END))
+
+    if price_type == "energy":
+        return __serialize_prices(
+            await energyzero.energy_prices(start_date=start, end_date=end)
+        )
+
+    return __serialize_prices(
+        await energyzero.gas_prices(start_date=start, end_date=end)
+    )
+
+
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up EnergyZero from a config entry."""
 
@@ -25,6 +82,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    async def get_prices(call: ServiceCall) -> ServiceResponse:
+        """Search prices."""
+        return await _get_prices(hass, call)
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_NAME,
+        get_prices,
+        schema=SERVICE_SCHEMA,
+        supports_response=SupportsResponse.ONLY,
+    )
+
     return True
 
 

homeassistant/components/energyzero/const.py

@@ -5,12 +5,29 @@ from datetime import timedelta
 import logging
 from typing import Final
 
+import voluptuous as vol
+
 DOMAIN: Final = "energyzero"
 LOGGER = logging.getLogger(__package__)
 SCAN_INTERVAL = timedelta(minutes=10)
 THRESHOLD_HOUR: Final = 14
 
+ATTR_TYPE: Final = "type"
+ATTR_START: Final = "start"
+ATTR_END: Final = "end"
+ATTR_INCL_VAT: Final = "incl_vat"
+
 SERVICE_TYPE_DEVICE_NAMES = {
     "today_energy": "Energy market price",
     "today_gas": "Gas market price",
 }
+SERVICE_NAME: Final = "get_prices"
+SERVICE_PRICE_TYPES: Final = ["energy", "gas"]
+SERVICE_SCHEMA: Final = vol.Schema(
+    {
+        vol.Required(ATTR_TYPE): vol.In(SERVICE_PRICE_TYPES),
+        vol.Optional(ATTR_START): str,
+        vol.Optional(ATTR_END): str,
+        vol.Optional(ATTR_INCL_VAT, default=True): bool,
+    }
+)

homeassistant/components/energyzero/services.yaml

@@ -0,0 +1,25 @@
+get_prices:
+  fields:
+    type:
+      required: true
+      example: "gas"
+      selector:
+        select:
+          options:
+            - "gas"
+            - "energy"
+    incl_vat:
+      required: false
+      example: false
+      selector:
+        boolean:
+    start:
+      required: false
+      example: "2023-01-01 00:00:00"
+      selector:
+        datetime:
+    end:
+      required: false
+      example: "2023-01-01 00:00:00"
+      selector:
+        datetime:

homeassistant/components/energyzero/strings.json

@@ -39,5 +39,29 @@
         "name": "Hours priced equal or lower than current - today"
       }
     }
+  },
+  "services": {
+    "get_prices": {
+      "name": "Get prices",
+      "description": "Request energy or gas prices from EnergyZero.",
+      "fields": {
+        "type": {
+          "name": "Type",
+          "description": "Type of prices to get, energy or gas."
+        },
+        "incl_vat": {
+          "name": "Including VAT",
+          "description": "Include VAT in the prices. Defaults to true if omitted."
+        },
+        "start": {
+          "name": "Start",
+          "description": "From which moment to get the prices. Defaults to today if omitted."
+        },
+        "end": {
+          "name": "End",
+          "description": "Until which moment to get the prices. Defaults to today if omitted."
+        }
+      }
+    }
   }
 }

tests/components/energyzero/conftest.py

@@ -32,25 +32,42 @@ def mock_config_entry() -> MockConfigEntry:
     )
 
 
+def apply_energyzero_mock(energyzero_mock):
+    """Apply mocks to EnergyZero client."""
+    client = energyzero_mock.return_value
+    client.energy_prices.return_value = Electricity.from_dict(
+        json.loads(load_fixture("today_energy.json", DOMAIN))
+    )
+    client.gas_prices.return_value = Gas.from_dict(
+        json.loads(load_fixture("today_gas.json", DOMAIN))
+    )
+    return client
+
+
 @pytest.fixture
 def mock_energyzero() -> Generator[MagicMock, None, None]:
     """Return a mocked EnergyZero client."""
     with patch(
         "homeassistant.components.energyzero.coordinator.EnergyZero", autospec=True
     ) as energyzero_mock:
-        client = energyzero_mock.return_value
+        yield apply_energyzero_mock(energyzero_mock)
-        client.energy_prices.return_value = Electricity.from_dict(
+
-            json.loads(load_fixture("today_energy.json", DOMAIN))
+
-        )
+@pytest.fixture
-        client.gas_prices.return_value = Gas.from_dict(
+def mock_energyzero_service() -> Generator[MagicMock, None, None]:
-            json.loads(load_fixture("today_gas.json", DOMAIN))
+    """Return a mocked EnergyZero client."""
-        )
+    with patch(
-        yield client
+        "homeassistant.components.energyzero.EnergyZero", autospec=True
+    ) as energyzero_mock:
+        yield apply_energyzero_mock(energyzero_mock)
 
 
 @pytest.fixture
 async def init_integration(
-    hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_energyzero: MagicMock
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_energyzero: MagicMock,
+    mock_energyzero_service: MagicMock,
 ) -> MockConfigEntry:
     """Set up the EnergyZero integration for testing."""
     mock_config_entry.add_to_hass(hass)

tests/components/energyzero/test_services.py

@@ -0,0 +1,51 @@
+"""Tests for the sensors provided by the EnergyZero integration."""
+
+import pytest
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.energyzero.const import DOMAIN, SERVICE_NAME
+from homeassistant.core import HomeAssistant
+
+pytestmark = [pytest.mark.freeze_time("2022-12-07 15:00:00")]
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_has_service(
+    hass: HomeAssistant,
+) -> None:
+    """Test the existence of the EnergyZero Service."""
+    assert hass.services.has_service(DOMAIN, SERVICE_NAME)
+
+
+@pytest.mark.usefixtures("init_integration")
+@pytest.mark.parametrize("price_type", [{"type": "gas"}, {"type": "energy"}])
+@pytest.mark.parametrize("incl_vat", [{"incl_vat": False}, {"incl_vat": True}, {}])
+@pytest.mark.parametrize(
+    "start", [{"start": "2023-01-01 00:00:00"}, {"start": "incorrect date"}, {}]
+)
+@pytest.mark.parametrize(
+    "end", [{"end": "2023-01-01 00:00:00"}, {"end": "incorrect date"}, {}]
+)
+async def test_service(
+    hass: HomeAssistant,
+    snapshot: SnapshotAssertion,
+    price_type: dict[str, str],
+    incl_vat: dict[str, bool],
+    start: dict[str, str],
+    end: dict[str, str],
+) -> None:
+    """Test the EnergyZero Service."""
+
+    data = price_type | incl_vat | start | end
+
+    try:
+        response = await hass.services.async_call(
+            DOMAIN,
+            SERVICE_NAME,
+            data,
+            blocking=True,
+            return_response=True,
+        )
+        assert response == snapshot
+    except ValueError as e:
+        assert e == snapshot