Add amberelectric price descriptors (#67981)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

homeassistant/components/amberelectric/coordinator.py

@@ -10,6 +10,7 @@ from amberelectric.model.actual_interval import ActualInterval
 from amberelectric.model.channel import ChannelType
 from amberelectric.model.current_interval import CurrentInterval
 from amberelectric.model.forecast_interval import ForecastInterval
+from amberelectric.model.interval import Descriptor
 
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@@ -44,6 +45,27 @@ def is_feed_in(interval: ActualInterval | CurrentInterval | ForecastInterval) ->
     return interval.channel_type == ChannelType.FEED_IN
 
 
+def normalize_descriptor(descriptor: Descriptor) -> 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."""
 
@@ -65,6 +87,7 @@ class AmberUpdateCoordinator(DataUpdateCoordinator):
 
         result: dict[str, dict[str, Any]] = {
             "current": {},
+            "descriptors": {},
             "forecasts": {},
             "grid": {},
         }
@@ -81,6 +104,7 @@ class AmberUpdateCoordinator(DataUpdateCoordinator):
             raise UpdateFailed("No general channel configured")
 
         result["current"]["general"] = general[0]
+        result["descriptors"]["general"] = normalize_descriptor(general[0].descriptor)
         result["forecasts"]["general"] = [
             interval for interval in forecasts if is_general(interval)
         ]
@@ -92,6 +116,9 @@ class AmberUpdateCoordinator(DataUpdateCoordinator):
         ]
         if controlled_load:
             result["current"]["controlled_load"] = controlled_load[0]
+            result["descriptors"]["controlled_load"] = normalize_descriptor(
+                controlled_load[0].descriptor
+            )
             result["forecasts"]["controlled_load"] = [
                 interval for interval in forecasts if is_controlled_load(interval)
             ]
@@ -99,6 +126,9 @@ class AmberUpdateCoordinator(DataUpdateCoordinator):
         feed_in = [interval for interval in current if is_feed_in(interval)]
         if feed_in:
             result["current"]["feed_in"] = feed_in[0]
+            result["descriptors"]["feed_in"] = normalize_descriptor(
+                feed_in[0].descriptor
+            )
             result["forecasts"]["feed_in"] = [
                 interval for interval in forecasts if is_feed_in(interval)
             ]

homeassistant/components/amberelectric/manifest.json

@@ -7,7 +7,7 @@
         "@madpilot"
     ],
     "requirements": [
-        "amberelectric==1.0.3"
+        "amberelectric==1.0.4"
     ],
     "iot_class": "cloud_polling",
     "loggers": ["amberelectric"]

homeassistant/components/amberelectric/sensor.py

@@ -26,7 +26,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import ATTRIBUTION, DOMAIN
-from .coordinator import AmberUpdateCoordinator
+from .coordinator import AmberUpdateCoordinator, normalize_descriptor
 
 ICONS = {
     "general": "mdi:transmission-tower",
@@ -160,6 +160,7 @@ class AmberForecastSensor(AmberSensor):
             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)
@@ -170,6 +171,15 @@ class AmberForecastSensor(AmberSensor):
         return data
 
 
+class AmberPriceDescriptorSensor(AmberSensor):
+    """Amber Price Descriptor Sensor."""
+
+    @property
+    def native_value(self) -> str | None:
+        """Return the current price descriptor."""
+        return self.coordinator.data[self.entity_description.key][self.channel_type]
+
+
 class AmberGridSensor(CoordinatorEntity, SensorEntity):
     """Sensor to show single grid specific values."""
 
@@ -214,6 +224,16 @@ async def async_setup_entry(
         )
         entities.append(AmberPriceSensor(coordinator, description, channel_type))
 
+    for channel_type in current:
+        description = SensorEntityDescription(
+            key="descriptors",
+            name=f"{entry.title} - {friendly_channel_type(channel_type)} Price Descriptor",
+            icon=ICONS[channel_type],
+        )
+        entities.append(
+            AmberPriceDescriptorSensor(coordinator, description, channel_type)
+        )
+
     for channel_type in forecasts:
         description = SensorEntityDescription(
             key="forecasts",

requirements_all.txt

@@ -280,7 +280,7 @@ alpha_vantage==2.3.1
 ambee==0.4.0
 
 # homeassistant.components.amberelectric
-amberelectric==1.0.3
+amberelectric==1.0.4
 
 # homeassistant.components.ambiclimate
 ambiclimate==0.2.1

requirements_test_all.txt

@@ -237,7 +237,7 @@ airtouch4pyapi==1.0.5
 ambee==0.4.0
 
 # homeassistant.components.amberelectric
-amberelectric==1.0.3
+amberelectric==1.0.4
 
 # homeassistant.components.ambiclimate
 ambiclimate==0.2.1

tests/components/amberelectric/helpers.py

@@ -6,7 +6,7 @@ from amberelectric.model.actual_interval import ActualInterval
 from amberelectric.model.channel import ChannelType
 from amberelectric.model.current_interval import CurrentInterval
 from amberelectric.model.forecast_interval import ForecastInterval
-from amberelectric.model.interval import SpikeStatus
+from amberelectric.model.interval import Descriptor, SpikeStatus
 from dateutil import parser
 
 
@@ -26,6 +26,7 @@ def generate_actual_interval(
         renewables=50,
         channel_type=channel_type.value,
         spike_status=SpikeStatus.NO_SPIKE.value,
+        descriptor=Descriptor.LOW.value,
     )
 
 
@@ -45,6 +46,7 @@ def generate_current_interval(
         renewables=50.6,
         channel_type=channel_type.value,
         spike_status=SpikeStatus.NO_SPIKE.value,
+        descriptor=Descriptor.EXTREMELY_LOW.value,
         estimate=True,
     )
 
@@ -65,6 +67,7 @@ def generate_forecast_interval(
         renewables=50,
         channel_type=channel_type.value,
         spike_status=SpikeStatus.NO_SPIKE.value,
+        descriptor=Descriptor.VERY_LOW.value,
         estimate=True,
     )
 

tests/components/amberelectric/test_binary_sensor.py

@@ -112,7 +112,7 @@ async def setup_spike(hass) -> AsyncGenerator:
 
 def test_no_spike_sensor(hass: HomeAssistant, setup_no_spike) -> None:
     """Testing the creation of the Amber renewables sensor."""
-    assert len(hass.states.async_all()) == 4
+    assert len(hass.states.async_all()) == 5
     sensor = hass.states.get("binary_sensor.mock_title_price_spike")
     assert sensor
     assert sensor.state == "off"
@@ -122,7 +122,7 @@ def test_no_spike_sensor(hass: HomeAssistant, setup_no_spike) -> None:
 
 def test_potential_spike_sensor(hass: HomeAssistant, setup_potential_spike) -> None:
     """Testing the creation of the Amber renewables sensor."""
-    assert len(hass.states.async_all()) == 4
+    assert len(hass.states.async_all()) == 5
     sensor = hass.states.get("binary_sensor.mock_title_price_spike")
     assert sensor
     assert sensor.state == "off"
@@ -132,7 +132,7 @@ def test_potential_spike_sensor(hass: HomeAssistant, setup_potential_spike) -> N
 
 def test_spike_sensor(hass: HomeAssistant, setup_spike) -> None:
     """Testing the creation of the Amber renewables sensor."""
-    assert len(hass.states.async_all()) == 4
+    assert len(hass.states.async_all()) == 5
     sensor = hass.states.get("binary_sensor.mock_title_price_spike")
     assert sensor
     assert sensor.state == "on"

tests/components/amberelectric/test_coordinator.py

@@ -7,12 +7,15 @@ from unittest.mock import Mock, patch
 from amberelectric import ApiException
 from amberelectric.model.channel import Channel, ChannelType
 from amberelectric.model.current_interval import CurrentInterval
-from amberelectric.model.interval import SpikeStatus
+from amberelectric.model.interval import Descriptor, SpikeStatus
 from amberelectric.model.site import Site
 from dateutil import parser
 import pytest
 
-from homeassistant.components.amberelectric.coordinator import AmberUpdateCoordinator
+from homeassistant.components.amberelectric.coordinator import (
+    AmberUpdateCoordinator,
+    normalize_descriptor,
+)
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import UpdateFailed
 
@@ -63,6 +66,18 @@ def mock_api_current_price() -> Generator:
         yield instance
 
 
+def test_normalize_descriptor() -> None:
+    """Test normalizing descriptors works correctly."""
+    assert normalize_descriptor(None) is None
+    assert normalize_descriptor(Descriptor.NEGATIVE) == "negative"
+    assert normalize_descriptor(Descriptor.EXTREMELY_LOW) == "extremely_low"
+    assert normalize_descriptor(Descriptor.VERY_LOW) == "very_low"
+    assert normalize_descriptor(Descriptor.LOW) == "low"
+    assert normalize_descriptor(Descriptor.NEUTRAL) == "neutral"
+    assert normalize_descriptor(Descriptor.HIGH) == "high"
+    assert normalize_descriptor(Descriptor.SPIKE) == "spike"
+
+
 async def test_fetch_general_site(hass: HomeAssistant, current_price_api: Mock) -> None:
     """Test fetching a site with only a general channel."""
 

tests/components/amberelectric/test_sensor.py

@@ -101,7 +101,7 @@ async def setup_general_and_feed_in(hass) -> AsyncGenerator:
 
 async def test_general_price_sensor(hass: HomeAssistant, setup_general: Mock) -> None:
     """Test the General Price sensor."""
-    assert len(hass.states.async_all()) == 4
+    assert len(hass.states.async_all()) == 5
     price = hass.states.get("sensor.mock_title_general_price")
     assert price
     assert price.state == "0.08"
@@ -140,7 +140,7 @@ async def test_general_and_controlled_load_price_sensor(
     hass: HomeAssistant, setup_general_and_controlled_load: Mock
 ) -> None:
     """Test the Controlled Price sensor."""
-    assert len(hass.states.async_all()) == 6
+    assert len(hass.states.async_all()) == 8
     price = hass.states.get("sensor.mock_title_controlled_load_price")
     assert price
     assert price.state == "0.08"
@@ -163,7 +163,7 @@ async def test_general_and_feed_in_price_sensor(
     hass: HomeAssistant, setup_general_and_feed_in: Mock
 ) -> None:
     """Test the Feed In sensor."""
-    assert len(hass.states.async_all()) == 6
+    assert len(hass.states.async_all()) == 8
     price = hass.states.get("sensor.mock_title_feed_in_price")
     assert price
     assert price.state == "-0.08"
@@ -186,7 +186,7 @@ async def test_general_forecast_sensor(
     hass: HomeAssistant, setup_general: Mock
 ) -> None:
     """Test the General Forecast sensor."""
-    assert len(hass.states.async_all()) == 4
+    assert len(hass.states.async_all()) == 5
     price = hass.states.get("sensor.mock_title_general_forecast")
     assert price
     assert price.state == "0.09"
@@ -204,6 +204,7 @@ async def test_general_forecast_sensor(
     assert first_forecast["end_time"] == "2021-09-21T09:00:00+10:00"
     assert first_forecast["renewables"] == 50
     assert first_forecast["spike_status"] == "none"
+    assert first_forecast["descriptor"] == "very_low"
 
     assert first_forecast.get("range_min") is None
     assert first_forecast.get("range_max") is None
@@ -228,7 +229,7 @@ async def test_controlled_load_forecast_sensor(
     hass: HomeAssistant, setup_general_and_controlled_load: Mock
 ) -> None:
     """Test the Controlled Load Forecast sensor."""
-    assert len(hass.states.async_all()) == 6
+    assert len(hass.states.async_all()) == 8
     price = hass.states.get("sensor.mock_title_controlled_load_forecast")
     assert price
     assert price.state == "0.09"
@@ -246,13 +247,14 @@ async def test_controlled_load_forecast_sensor(
     assert first_forecast["end_time"] == "2021-09-21T09:00:00+10:00"
     assert first_forecast["renewables"] == 50
     assert first_forecast["spike_status"] == "none"
+    assert first_forecast["descriptor"] == "very_low"
 
 
 async def test_feed_in_forecast_sensor(
     hass: HomeAssistant, setup_general_and_feed_in: Mock
 ) -> None:
     """Test the Feed In Forecast sensor."""
-    assert len(hass.states.async_all()) == 6
+    assert len(hass.states.async_all()) == 8
     price = hass.states.get("sensor.mock_title_feed_in_forecast")
     assert price
     assert price.state == "-0.09"
@@ -270,11 +272,42 @@ async def test_feed_in_forecast_sensor(
     assert first_forecast["end_time"] == "2021-09-21T09:00:00+10:00"
     assert first_forecast["renewables"] == 50
     assert first_forecast["spike_status"] == "none"
+    assert first_forecast["descriptor"] == "very_low"
 
 
 def test_renewable_sensor(hass: HomeAssistant, setup_general) -> None:
     """Testing the creation of the Amber renewables sensor."""
-    assert len(hass.states.async_all()) == 4
+    assert len(hass.states.async_all()) == 5
     sensor = hass.states.get("sensor.mock_title_renewables")
     assert sensor
     assert sensor.state == "51"
+
+
+def test_general_price_descriptor_descriptor_sensor(
+    hass: HomeAssistant, setup_general: Mock
+) -> None:
+    """Test the General Price Descriptor sensor."""
+    assert len(hass.states.async_all()) == 5
+    price = hass.states.get("sensor.mock_title_general_price_descriptor")
+    assert price
+    assert price.state == "extremely_low"
+
+
+def test_general_and_controlled_load_price_descriptor_sensor(
+    hass: HomeAssistant, setup_general_and_controlled_load: Mock
+) -> None:
+    """Test the Controlled Price Descriptor sensor."""
+    assert len(hass.states.async_all()) == 8
+    price = hass.states.get("sensor.mock_title_controlled_load_price_descriptor")
+    assert price
+    assert price.state == "extremely_low"
+
+
+def test_general_and_feed_in_price_descriptor_sensor(
+    hass: HomeAssistant, setup_general_and_feed_in: Mock
+) -> None:
+    """Test the Feed In Price Descriptor sensor."""
+    assert len(hass.states.async_all()) == 8
+    price = hass.states.get("sensor.mock_title_feed_in_price_descriptor")
+    assert price
+    assert price.state == "extremely_low"