Adding price spike binary sensor to the Amber electric integration (#56736)

homeassistant/components/amberelectric/binary_sensor.py

@@ -0,0 +1,88 @@
+"""Amber Electric Binary Sensor definitions."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import ATTR_ATTRIBUTION
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import ATTRIBUTION, DOMAIN
+from .coordinator import AmberUpdateCoordinator
+
+PRICE_SPIKE_ICONS = {
+    "none": "mdi:power-plug",
+    "potential": "mdi:power-plug-outline",
+    "spike": "mdi:power-plug-off",
+}
+
+
+class AmberPriceGridSensor(CoordinatorEntity, BinarySensorEntity):
+    """Sensor to show single grid binary values."""
+
+    def __init__(
+        self,
+        coordinator: AmberUpdateCoordinator,
+        description: BinarySensorEntityDescription,
+    ) -> None:
+        """Initialize the Sensor."""
+        super().__init__(coordinator)
+        self.site_id = coordinator.site_id
+        self.entity_description = description
+        self._attr_unique_id = f"{coordinator.site_id}-{description.key}"
+        self._attr_device_state_attributes = {ATTR_ATTRIBUTION: ATTRIBUTION}
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if the binary sensor is on."""
+        return self.coordinator.data["grid"][self.entity_description.key]
+
+
+class AmberPriceSpikeBinarySensor(AmberPriceGridSensor):
+    """Sensor to show single grid binary values."""
+
+    @property
+    def icon(self):
+        """Return the sensor icon."""
+        status = self.coordinator.data["grid"]["price_spike"]
+        return PRICE_SPIKE_ICONS[status]
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if the binary sensor is on."""
+        return self.coordinator.data["grid"]["price_spike"] == "spike"
+
+    @property
+    def device_state_attributes(self) -> Mapping[str, Any] | None:
+        """Return additional pieces of information about the price spike."""
+
+        spike_status = self.coordinator.data["grid"]["price_spike"]
+        return {
+            "spike_status": spike_status,
+            ATTR_ATTRIBUTION: ATTRIBUTION,
+        }
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up a config entry."""
+    coordinator: AmberUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+
+    entities: list = []
+    price_spike_description = BinarySensorEntityDescription(
+        key="price_spike",
+        name=f"{entry.title} - Price Spike",
+    )
+    entities.append(AmberPriceSpikeBinarySensor(coordinator, price_spike_description))
+    async_add_entities(entities)

homeassistant/components/amberelectric/const.py

@@ -7,5 +7,7 @@ CONF_SITE_NAME = "site_name"
 CONF_SITE_ID = "site_id"
 CONF_SITE_NMI = "site_nmi"
 
+ATTRIBUTION = "Data provided by Amber Electric"
+
 LOGGER = logging.getLogger(__package__)
-PLATFORMS = ["sensor"]
+PLATFORMS = ["sensor", "binary_sensor"]

homeassistant/components/amberelectric/coordinator.py

@@ -85,6 +85,7 @@ class AmberUpdateCoordinator(DataUpdateCoordinator):
             interval for interval in forecasts if is_general(interval)
         ]
         result["grid"]["renewables"] = round(general[0].renewables)
+        result["grid"]["price_spike"] = general[0].spike_status.value
 
         controlled_load = [
             interval for interval in current if is_controlled_load(interval)

homeassistant/components/amberelectric/sensor.py

@@ -25,11 +25,9 @@ from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from .const import DOMAIN
+from .const import ATTRIBUTION, DOMAIN
 from .coordinator import AmberUpdateCoordinator
 
-ATTRIBUTION = "Data provided by Amber Electric"
-
 ICONS = {
     "general": "mdi:transmission-tower",
     "controlled_load": "mdi:clock-outline",
@@ -63,9 +61,6 @@ class AmberSensor(CoordinatorEntity, SensorEntity):
         self.entity_description = description
         self.channel_type = channel_type
 
-    @property
-    def unique_id(self) -> None:
-        """Return a unique id for each sensors."""
         self._attr_unique_id = (
             f"{self.site_id}-{self.entity_description.key}-{self.channel_type}"
         )
@@ -119,9 +114,11 @@ class AmberForecastSensor(AmberSensor):
     @property
     def native_value(self) -> str | None:
         """Return the first forecast price in $/kWh."""
-        intervals = self.coordinator.data[self.entity_description.key][
+        intervals = self.coordinator.data[self.entity_description.key].get(
             self.channel_type
-        ]
+        )
+        if not intervals:
+            return None
         interval = intervals[0]
 
         if interval.channel_type == ChannelType.FEED_IN:
@@ -131,9 +128,12 @@ class AmberForecastSensor(AmberSensor):
     @property
     def device_state_attributes(self) -> Mapping[str, Any] | None:
         """Return additional pieces of information about the price."""
-        intervals = self.coordinator.data[self.entity_description.key][
+        intervals = self.coordinator.data[self.entity_description.key].get(
             self.channel_type
-        ]
+        )
+
+        if not intervals:
+            return None
 
         data = {
             "forecasts": [],
@@ -179,11 +179,6 @@ class AmberGridSensor(CoordinatorEntity, SensorEntity):
         self._attr_device_state_attributes = {ATTR_ATTRIBUTION: ATTRIBUTION}
         self._attr_unique_id = f"{coordinator.site_id}-{description.key}"
 
-    @property
-    def unique_id(self) -> None:
-        """Return a unique id for each sensors."""
-        self._attr_unique_id = f"{self.site_id}-{self.entity_description.key}"
-
     @property
     def native_value(self) -> str | None:
         """Return the value of the sensor."""

tests/components/amberelectric/test_binary_sensor.py

@@ -0,0 +1,140 @@
+"""Test the Amber Electric Sensors."""
+from __future__ import annotations
+
+from typing import AsyncGenerator
+from unittest.mock import Mock, patch
+
+from amberelectric.model.channel import ChannelType
+from amberelectric.model.current_interval import CurrentInterval
+from amberelectric.model.interval import SpikeStatus
+from dateutil import parser
+import pytest
+
+from homeassistant.components.amberelectric.const import (
+    CONF_API_TOKEN,
+    CONF_SITE_ID,
+    CONF_SITE_NAME,
+    DOMAIN,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry
+from tests.components.amberelectric.helpers import (
+    GENERAL_CHANNEL,
+    GENERAL_ONLY_SITE_ID,
+    generate_current_interval,
+)
+
+MOCK_API_TOKEN = "psk_0000000000000000"
+
+
+@pytest.fixture
+async def setup_no_spike(hass) -> AsyncGenerator:
+    """Set up general channel."""
+    MockConfigEntry(
+        domain="amberelectric",
+        data={
+            CONF_SITE_NAME: "mock_title",
+            CONF_API_TOKEN: MOCK_API_TOKEN,
+            CONF_SITE_ID: GENERAL_ONLY_SITE_ID,
+        },
+    ).add_to_hass(hass)
+
+    instance = Mock()
+    with patch(
+        "amberelectric.api.AmberApi.create",
+        return_value=instance,
+    ) as mock_update:
+        instance.get_current_price = Mock(return_value=GENERAL_CHANNEL)
+        assert await async_setup_component(hass, DOMAIN, {})
+        await hass.async_block_till_done()
+        yield mock_update.return_value
+
+
+@pytest.fixture
+async def setup_potential_spike(hass) -> AsyncGenerator:
+    """Set up general channel."""
+    MockConfigEntry(
+        domain="amberelectric",
+        data={
+            CONF_SITE_NAME: "mock_title",
+            CONF_API_TOKEN: MOCK_API_TOKEN,
+            CONF_SITE_ID: GENERAL_ONLY_SITE_ID,
+        },
+    ).add_to_hass(hass)
+
+    instance = Mock()
+    with patch(
+        "amberelectric.api.AmberApi.create",
+        return_value=instance,
+    ) as mock_update:
+        general_channel: list[CurrentInterval] = [
+            generate_current_interval(
+                ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00")
+            ),
+        ]
+        general_channel[0].spike_status = SpikeStatus.POTENTIAL
+        instance.get_current_price = Mock(return_value=general_channel)
+        assert await async_setup_component(hass, DOMAIN, {})
+        await hass.async_block_till_done()
+        yield mock_update.return_value
+
+
+@pytest.fixture
+async def setup_spike(hass) -> AsyncGenerator:
+    """Set up general channel."""
+    MockConfigEntry(
+        domain="amberelectric",
+        data={
+            CONF_SITE_NAME: "mock_title",
+            CONF_API_TOKEN: MOCK_API_TOKEN,
+            CONF_SITE_ID: GENERAL_ONLY_SITE_ID,
+        },
+    ).add_to_hass(hass)
+
+    instance = Mock()
+    with patch(
+        "amberelectric.api.AmberApi.create",
+        return_value=instance,
+    ) as mock_update:
+        general_channel: list[CurrentInterval] = [
+            generate_current_interval(
+                ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00")
+            ),
+        ]
+        general_channel[0].spike_status = SpikeStatus.SPIKE
+        instance.get_current_price = Mock(return_value=general_channel)
+        assert await async_setup_component(hass, DOMAIN, {})
+        await hass.async_block_till_done()
+        yield mock_update.return_value
+
+
+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
+    sensor = hass.states.get("binary_sensor.mock_title_price_spike")
+    assert sensor
+    assert sensor.state == "off"
+    assert sensor.attributes["icon"] == "mdi:power-plug"
+    assert sensor.attributes["spike_status"] == "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
+    sensor = hass.states.get("binary_sensor.mock_title_price_spike")
+    assert sensor
+    assert sensor.state == "off"
+    assert sensor.attributes["icon"] == "mdi:power-plug-outline"
+    assert sensor.attributes["spike_status"] == "potential"
+
+
+def test_spike_sensor(hass: HomeAssistant, setup_spike) -> None:
+    """Testing the creation of the Amber renewables sensor."""
+    assert len(hass.states.async_all()) == 4
+    sensor = hass.states.get("binary_sensor.mock_title_price_spike")
+    assert sensor
+    assert sensor.state == "on"
+    assert sensor.attributes["icon"] == "mdi:power-plug-off"
+    assert sensor.attributes["spike_status"] == "spike"

tests/components/amberelectric/test_coordinator.py

@@ -1,10 +1,15 @@
 """Tests for the Amber Electric Data Coordinator."""
+from __future__ import annotations
+
 from typing import Generator
 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.site import Site
+from dateutil import parser
 import pytest
 
 from homeassistant.components.amberelectric.coordinator import AmberUpdateCoordinator
@@ -18,6 +23,7 @@ from tests.components.amberelectric.helpers import (
     GENERAL_AND_FEED_IN_SITE_ID,
     GENERAL_CHANNEL,
     GENERAL_ONLY_SITE_ID,
+    generate_current_interval,
 )
 
 
@@ -79,6 +85,7 @@ async def test_fetch_general_site(hass: HomeAssistant, current_price_api: Mock)
     assert result["current"].get("feed_in") is None
     assert result["forecasts"].get("feed_in") is None
     assert result["grid"]["renewables"] == round(GENERAL_CHANNEL[0].renewables)
+    assert result["grid"]["price_spike"] == "none"
 
 
 async def test_fetch_no_general_site(
@@ -134,6 +141,7 @@ async def test_fetch_api_error(hass: HomeAssistant, current_price_api: Mock) ->
     assert result["current"].get("feed_in") is None
     assert result["forecasts"].get("feed_in") is None
     assert result["grid"]["renewables"] == round(GENERAL_CHANNEL[0].renewables)
+    assert result["grid"]["price_spike"] == "none"
 
 
 async def test_fetch_general_and_controlled_load_site(
@@ -168,6 +176,7 @@ async def test_fetch_general_and_controlled_load_site(
     assert result["current"].get("feed_in") is None
     assert result["forecasts"].get("feed_in") is None
     assert result["grid"]["renewables"] == round(GENERAL_CHANNEL[0].renewables)
+    assert result["grid"]["price_spike"] == "none"
 
 
 async def test_fetch_general_and_feed_in_site(
@@ -200,3 +209,36 @@ async def test_fetch_general_and_feed_in_site(
         FEED_IN_CHANNEL[3],
     ]
     assert result["grid"]["renewables"] == round(GENERAL_CHANNEL[0].renewables)
+    assert result["grid"]["price_spike"] == "none"
+
+
+async def test_fetch_potential_spike(
+    hass: HomeAssistant, current_price_api: Mock
+) -> None:
+    """Test fetching a site with only a general channel."""
+
+    general_channel: list[CurrentInterval] = [
+        generate_current_interval(
+            ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00")
+        ),
+    ]
+    general_channel[0].spike_status = SpikeStatus.POTENTIAL
+    current_price_api.get_current_price.return_value = general_channel
+    data_service = AmberUpdateCoordinator(hass, current_price_api, GENERAL_ONLY_SITE_ID)
+    result = await data_service._async_update_data()
+    assert result["grid"]["price_spike"] == "potential"
+
+
+async def test_fetch_spike(hass: HomeAssistant, current_price_api: Mock) -> None:
+    """Test fetching a site with only a general channel."""
+
+    general_channel: list[CurrentInterval] = [
+        generate_current_interval(
+            ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00")
+        ),
+    ]
+    general_channel[0].spike_status = SpikeStatus.SPIKE
+    current_price_api.get_current_price.return_value = general_channel
+    data_service = AmberUpdateCoordinator(hass, current_price_api, GENERAL_ONLY_SITE_ID)
+    result = await data_service._async_update_data()
+    assert result["grid"]["price_spike"] == "spike"

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()) == 3
+    assert len(hass.states.async_all()) == 4
     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()) == 5
+    assert len(hass.states.async_all()) == 6
     print(hass.states)
     price = hass.states.get("sensor.mock_title_controlled_load_price")
     assert price
@@ -164,7 +164,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()) == 5
+    assert len(hass.states.async_all()) == 6
     print(hass.states)
     price = hass.states.get("sensor.mock_title_feed_in_price")
     assert price
@@ -188,7 +188,7 @@ async def test_general_forecast_sensor(
     hass: HomeAssistant, setup_general: Mock
 ) -> None:
     """Test the General Forecast sensor."""
-    assert len(hass.states.async_all()) == 3
+    assert len(hass.states.async_all()) == 4
     price = hass.states.get("sensor.mock_title_general_forecast")
     assert price
     assert price.state == "0.09"
@@ -230,7 +230,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()) == 5
+    assert len(hass.states.async_all()) == 6
     price = hass.states.get("sensor.mock_title_controlled_load_forecast")
     assert price
     assert price.state == "0.09"
@@ -254,7 +254,7 @@ 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()) == 5
+    assert len(hass.states.async_all()) == 6
     price = hass.states.get("sensor.mock_title_feed_in_forecast")
     assert price
     assert price.state == "-0.09"
@@ -276,7 +276,7 @@ async def test_feed_in_forecast_sensor(
 
 def test_renewable_sensor(hass: HomeAssistant, setup_general) -> None:
     """Testing the creation of the Amber renewables sensor."""
-    assert len(hass.states.async_all()) == 3
+    assert len(hass.states.async_all()) == 4
     sensor = hass.states.get("sensor.mock_title_renewables")
     assert sensor
     assert sensor.state == "51"