Add forecast service to amberelectric (home-assistant#144848)

Co-authored-by: G Johansson <[email protected]>

Author: madpilot

homeassistant/components/amberelectric/__init__.py

@@ -2,11 +2,22 @@
 
 import amberelectric
 
+from homeassistant.components.sensor import ConfigType
 from homeassistant.const import CONF_API_TOKEN
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_validation as cv
 
-from .const import CONF_SITE_ID, PLATFORMS
+from .const import CONF_SITE_ID, DOMAIN, PLATFORMS
 from .coordinator import AmberConfigEntry, AmberUpdateCoordinator
+from .services import setup_services
+
+CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the Amber component."""
+    setup_services(hass)
+    return True
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: AmberConfigEntry) -> bool:

homeassistant/components/amberelectric/const.py

@@ -1,14 +1,24 @@
 """Amber Electric Constants."""
 
 import logging
+from typing import Final
 
 from homeassistant.const import Platform
 
-DOMAIN = "amberelectric"
+DOMAIN: Final = "amberelectric"
 CONF_SITE_NAME = "site_name"
 CONF_SITE_ID = "site_id"
 
+ATTR_CONFIG_ENTRY_ID = "config_entry_id"
+ATTR_CHANNEL_TYPE = "channel_type"
+
 ATTRIBUTION = "Data provided by Amber Electric"
 
 LOGGER = logging.getLogger(__package__)
 PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
+
+SERVICE_GET_FORECASTS = "get_forecasts"
+
+GENERAL_CHANNEL = "general"
+CONTROLLED_LOAD_CHANNEL = "controlled_load"
+FEED_IN_CHANNEL = "feed_in"

homeassistant/components/amberelectric/coordinator.py

@@ -10,14 +10,14 @@
 from amberelectric.models.channel import ChannelType
 from amberelectric.models.current_interval import CurrentInterval
 from amberelectric.models.forecast_interval import ForecastInterval
-from amberelectric.models.price_descriptor import PriceDescriptor
 from amberelectric.rest import ApiException
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import LOGGER
+from .helpers import normalize_descriptor
 
 type AmberConfigEntry = ConfigEntry[AmberUpdateCoordinator]
 
@@ -49,27 +49,6 @@ def is_feed_in(interval: ActualInterval | CurrentInterval | ForecastInterval) ->
     return interval.channel_type == ChannelType.FEEDIN
 
 
-def normalize_descriptor(descriptor: PriceDescriptor | None) -> str | None:
-    """Return the snake case versions of descriptor names. Returns None if the name is not recognized."""
-    if descriptor is None:
-        return None
-    if descriptor.value == "spike":
-        return "spike"
-    if descriptor.value == "high":
-        return "high"
-    if descriptor.value == "neutral":
-        return "neutral"
-    if descriptor.value == "low":
-        return "low"
-    if descriptor.value == "veryLow":
-        return "very_low"
-    if descriptor.value == "extremelyLow":
-        return "extremely_low"
-    if descriptor.value == "negative":
-        return "negative"
-    return None
-
-
 class AmberUpdateCoordinator(DataUpdateCoordinator):
     """AmberUpdateCoordinator - In charge of downloading the data for a site, which all the sensors read."""
 
@@ -103,7 +82,7 @@ def update_price_data(self) -> dict[str, dict[str, Any]]:
             "grid": {},
         }
         try:
-            data = self._api.get_current_prices(self.site_id, next=48)
+            data = self._api.get_current_prices(self.site_id, next=288)
             intervals = [interval.actual_instance for interval in data]
         except ApiException as api_exception:
             raise UpdateFailed("Missing price data, skipping update") from api_exception

homeassistant/components/amberelectric/helpers.py

@@ -0,0 +1,25 @@
+"""Formatting helpers used to convert things."""
+
+from amberelectric.models.price_descriptor import PriceDescriptor
+
+DESCRIPTOR_MAP: dict[str, str] = {
+    PriceDescriptor.SPIKE: "spike",
+    PriceDescriptor.HIGH: "high",
+    PriceDescriptor.NEUTRAL: "neutral",
+    PriceDescriptor.LOW: "low",
+    PriceDescriptor.VERYLOW: "very_low",
+    PriceDescriptor.EXTREMELYLOW: "extremely_low",
+    PriceDescriptor.NEGATIVE: "negative",
+}
+
+
+def normalize_descriptor(descriptor: PriceDescriptor | None) -> str | None:
+    """Return the snake case versions of descriptor names. Returns None if the name is not recognized."""
+    if descriptor in DESCRIPTOR_MAP:
+        return DESCRIPTOR_MAP[descriptor]
+    return None
+
+
+def format_cents_to_dollars(cents: float) -> float:
+    """Return a formatted conversion from cents to dollars."""
+    return round(cents / 100, 2)

homeassistant/components/amberelectric/icons.json

@@ -22,5 +22,10 @@
         }
       }
     }
+  },
+  "services": {
+    "get_forecasts": {
+      "service": "mdi:transmission-tower"
+    }
   }
 }

homeassistant/components/amberelectric/sensor.py

@@ -23,16 +23,12 @@
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import ATTRIBUTION
-from .coordinator import AmberConfigEntry, AmberUpdateCoordinator, normalize_descriptor
+from .coordinator import AmberConfigEntry, AmberUpdateCoordinator
+from .helpers import format_cents_to_dollars, normalize_descriptor
 
 UNIT = f"{CURRENCY_DOLLAR}/{UnitOfEnergy.KILO_WATT_HOUR}"
 
 
-def format_cents_to_dollars(cents: float) -> float:
-    """Return a formatted conversion from cents to dollars."""
-    return round(cents / 100, 2)
-
-
 def friendly_channel_type(channel_type: str) -> str:
     """Return a human readable version of the channel type."""
     if channel_type == "controlled_load":

homeassistant/components/amberelectric/services.py

@@ -0,0 +1,121 @@
+"""Amber Electric Service class."""
+
+from amberelectric.models.channel import ChannelType
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import (
+    HomeAssistant,
+    ServiceCall,
+    ServiceResponse,
+    SupportsResponse,
+)
+from homeassistant.exceptions import ServiceValidationError
+from homeassistant.helpers.selector import ConfigEntrySelector
+from homeassistant.util.json import JsonValueType
+
+from .const import (
+    ATTR_CHANNEL_TYPE,
+    ATTR_CONFIG_ENTRY_ID,
+    CONTROLLED_LOAD_CHANNEL,
+    DOMAIN,
+    FEED_IN_CHANNEL,
+    GENERAL_CHANNEL,
+    SERVICE_GET_FORECASTS,
+)
+from .coordinator import AmberConfigEntry
+from .helpers import format_cents_to_dollars, normalize_descriptor
+
+GET_FORECASTS_SCHEMA = vol.Schema(
+    {
+        ATTR_CONFIG_ENTRY_ID: ConfigEntrySelector({"integration": DOMAIN}),
+        ATTR_CHANNEL_TYPE: vol.In(
+            [GENERAL_CHANNEL, CONTROLLED_LOAD_CHANNEL, FEED_IN_CHANNEL]
+        ),
+    }
+)
+
+
+def async_get_entry(hass: HomeAssistant, config_entry_id: str) -> AmberConfigEntry:
+    """Get the Amber config entry."""
+    if not (entry := hass.config_entries.async_get_entry(config_entry_id)):
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="integration_not_found",
+            translation_placeholders={"target": config_entry_id},
+        )
+    if entry.state is not ConfigEntryState.LOADED:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="not_loaded",
+            translation_placeholders={"target": entry.title},
+        )
+    return entry
+
+
+def get_forecasts(channel_type: str, data: dict) -> list[JsonValueType]:
+    """Return an array of forecasts."""
+    results: list[JsonValueType] = []
+
+    if channel_type not in data["forecasts"]:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="channel_not_found",
+            translation_placeholders={"channel_type": channel_type},
+        )
+
+    intervals = data["forecasts"][channel_type]
+
+    for interval in intervals:
+        datum = {}
+        datum["duration"] = interval.duration
+        datum["date"] = interval.var_date.isoformat()
+        datum["nem_date"] = interval.nem_time.isoformat()
+        datum["per_kwh"] = format_cents_to_dollars(interval.per_kwh)
+        if interval.channel_type == ChannelType.FEEDIN:
+            datum["per_kwh"] = datum["per_kwh"] * -1
+        datum["spot_per_kwh"] = format_cents_to_dollars(interval.spot_per_kwh)
+        datum["start_time"] = interval.start_time.isoformat()
+        datum["end_time"] = interval.end_time.isoformat()
+        datum["renewables"] = round(interval.renewables)
+        datum["spike_status"] = interval.spike_status.value
+        datum["descriptor"] = normalize_descriptor(interval.descriptor)
+
+        if interval.range is not None:
+            datum["range_min"] = format_cents_to_dollars(interval.range.min)
+            datum["range_max"] = format_cents_to_dollars(interval.range.max)
+
+        if interval.advanced_price is not None:
+            multiplier = -1 if interval.channel_type == ChannelType.FEEDIN else 1
+            datum["advanced_price_low"] = multiplier * format_cents_to_dollars(
+                interval.advanced_price.low
+            )
+            datum["advanced_price_predicted"] = multiplier * format_cents_to_dollars(
+                interval.advanced_price.predicted
+            )
+            datum["advanced_price_high"] = multiplier * format_cents_to_dollars(
+                interval.advanced_price.high
+            )
+
+        results.append(datum)
+
+    return results
+
+
+def setup_services(hass: HomeAssistant) -> None:
+    """Set up the services for the Amber integration."""
+
+    async def handle_get_forecasts(call: ServiceCall) -> ServiceResponse:
+        channel_type = call.data[ATTR_CHANNEL_TYPE]
+        entry = async_get_entry(hass, call.data[ATTR_CONFIG_ENTRY_ID])
+        coordinator = entry.runtime_data
+        forecasts = get_forecasts(channel_type, coordinator.data)
+        return {"forecasts": forecasts}
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_GET_FORECASTS,
+        handle_get_forecasts,
+        GET_FORECASTS_SCHEMA,
+        supports_response=SupportsResponse.ONLY,
+    )

homeassistant/components/amberelectric/services.yaml

@@ -0,0 +1,16 @@
+get_forecasts:
+  fields:
+    config_entry_id:
+      required: true
+      selector:
+        config_entry:
+          integration: amberelectric
+    channel_type:
+      required: true
+      selector:
+        select:
+          options:
+            - general
+            - controlled_load
+            - feed_in
+          translation_key: channel_type

homeassistant/components/amberelectric/strings.json

@@ -1,25 +1,61 @@
 {
   "config": {
+    "error": {
+      "invalid_api_token": "[%key:common::config_flow::error::invalid_api_key%]",
+      "no_site": "No site provided",
+      "unknown_error": "[%key:common::config_flow::error::unknown%]"
+    },
     "step": {
+      "site": {
+        "data": {
+          "site_id": "Site NMI",
+          "site_name": "Site name"
+        },
+        "description": "Select the NMI of the site you would like to add"
+      },
       "user": {
         "data": {
           "api_token": "[%key:common::config_flow::data::api_token%]",
           "site_id": "Site ID"
         },
         "description": "Go to {api_url} to generate an API key"
-      },
-      "site": {
-        "data": {
-          "site_id": "Site NMI",
-          "site_name": "Site Name"
+      }
+    }
+  },
+  "services": {
+    "get_forecasts": {
+      "name": "Get price forecasts",
+      "description": "Retrieves price forecasts from Amber Electric for a site.",
+      "fields": {
+        "config_entry_id": {
+          "description": "The config entry of the site to get forecasts for.",
+          "name": "Config entry"
         },
-        "description": "Select the NMI of the site you would like to add"
+        "channel_type": {
+          "name": "Channel type",
+          "description": "The channel to get forecasts for."
+        }
       }
+    }
+  },
+  "exceptions": {
+    "integration_not_found": {
+      "message": "Config entry \"{target}\" not found in registry."
     },
-    "error": {
-      "invalid_api_token": "[%key:common::config_flow::error::invalid_api_key%]",
-      "no_site": "No site provided",
-      "unknown_error": "[%key:common::config_flow::error::unknown%]"
+    "not_loaded": {
+      "message": "{target} is not loaded."
+    },
+    "channel_not_found": {
+      "message": "There is no {channel_type} channel at this site."
+    }
+  },
+  "selector": {
+    "channel_type": {
+      "options": {
+        "general": "General",
+        "controlled_load": "Controlled load",
+        "feed_in": "Feed-in"
+      }
     }
   }
 }

tests/components/amberelectric/__init__.py

@@ -1 +1,13 @@
 """Tests for the amberelectric integration."""
+
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
+    """Fixture for setting up the component."""
+    config_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()