jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 09:47:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Amberelectric (#56448)

Browse Source 
* Add Amber Electric integration

* Linting

* Fixing some type hinting

* Adding docstrings

* Removing files that shouldn't have been changed

* Splitting out test helpers

* Testing the price sensor

* Testing Controlled load and feed in channels

* Refactoring mocks

* switching state for native_value and unit_of_measurement for native_unit_of_measurement

* Fixing docstrings

* Fixing requiremennts_all.txt

* isort fixes

* Fixing pylint errors

* Omitting __init__.py from test coverage

* Add missing config_flow tests

* Adding more sensor tests

* Applying suggested changes to __init.py__

* Refactor coordinator to return the data object with all of the relevent data already setup

* Another coordinator refactor - Better use the dictionary for when we build the sensors

* Removing first function

* Refactoring sensor files to use entity descriptions, remove factory

* Rounding renewable percentage, return icons correctly

* Cleaning up translation strings

* Fixing relative path, removing TODO

* Coordintator tests now accept new (more accurate) fixtures

* Using a description placeholder

* Putting missing translations strings back in

* tighten up the no site error logic - self._site_id should never be None at the point of loading async_step_site

* Removing DEVICE_CLASS, replacing the units with AUD/kWh

* Settings _attr_unique_id

* Removing icon function (it's already the default)

* Apply suggestions from code review

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>

* Adding strings.json

* Tighter wrapping for try/except

* Generating translations

* Removing update_method - not needed as it's being overriden

* Apply suggestions from code review

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>

* Fixing tests

* Add missing description placeholder

* Fix warning

* changing name from update to update_data to match async_update_data

* renaming [async_]update_data => [async_]update_price_data to avoid confusion

* Creating too man renewable sensors

* Override update method

* Coordinator tests use _async_update_data

* Using $/kWh as the units

* Using isinstance instead of __class__ test. Removing a zero len check

* Asserting self._sites in second step

* Linting

* Remove useless tests

Co-authored-by: jan iversen <jancasacondor@gmail.com>
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Myles Eftos 2021-09-28 17:03:51 +10:00 committed by GitHub
parent f93539ef4c
commit 412ecacca3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1326 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -51,6 +51,7 @@ omit =
homeassistant/components/alarmdecoder/sensor.py
homeassistant/components/alpha_vantage/sensor.py
homeassistant/components/amazon_polly/*
homeassistant/components/amberelectric/__init__.py
homeassistant/components/ambiclimate/climate.py
homeassistant/components/ambient_station/*
homeassistant/components/amcrest/*

1
CODEOWNERS
View File

@ -37,6 +37,7 @@ homeassistant/components/alexa/* @home-assistant/cloud @ochlocracy
homeassistant/components/almond/* @gcampax @balloob
homeassistant/components/alpha_vantage/* @fabaff
homeassistant/components/ambee/* @frenck
homeassistant/components/amberelectric/* @madpilot
homeassistant/components/ambiclimate/* @danielhiversen
homeassistant/components/ambient_station/* @bachya
homeassistant/components/amcrest/* @flacjacket

32
homeassistant/components/amberelectric/__init__.py Normal file
View File

@ -0,0 +1,32 @@
"""Support for Amber Electric."""
from amberelectric import Configuration
from amberelectric.api import amber_api
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import CONF_API_TOKEN, CONF_SITE_ID, DOMAIN, PLATFORMS
from .coordinator import AmberUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Amber Electric from a config entry."""
configuration = Configuration(access_token=entry.data[CONF_API_TOKEN])
api_instance = amber_api.AmberApi.create(configuration)
site_id = entry.data[CONF_SITE_ID]
coordinator = AmberUpdateCoordinator(hass, api_instance, site_id)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

120
homeassistant/components/amberelectric/config_flow.py Normal file
View File

@ -0,0 +1,120 @@
"""Config flow for the Amber Electric integration."""
from __future__ import annotations
from typing import Any
import amberelectric
from amberelectric.api import amber_api
from amberelectric.model.site import Site
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_TOKEN
from .const import CONF_SITE_ID, CONF_SITE_NAME, CONF_SITE_NMI, DOMAIN
API_URL = "https://app.amber.com.au/developers"
class AmberElectricConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._errors: dict[str, str] = {}
self._sites: list[Site] | None = None
self._api_token: str | None = None
def _fetch_sites(self, token: str) -> list[Site] | None:
configuration = amberelectric.Configuration(access_token=token)
api = amber_api.AmberApi.create(configuration)
try:
sites = api.get_sites()
if len(sites) == 0:
self._errors[CONF_API_TOKEN] = "no_site"
return None
return sites
except amberelectric.ApiException as api_exception:
if api_exception.status == 403:
self._errors[CONF_API_TOKEN] = "invalid_api_token"
else:
self._errors[CONF_API_TOKEN] = "unknown_error"
return None
async def async_step_user(self, user_input: dict[str, Any] | None = None):
"""Step when user initializes a integration."""
self._errors = {}
self._sites = None
self._api_token = None
if user_input is not None:
token = user_input[CONF_API_TOKEN]
self._sites = await self.hass.async_add_executor_job(
self._fetch_sites, token
)
if self._sites is not None:
self._api_token = token
return await self.async_step_site()
else:
user_input = {CONF_API_TOKEN: ""}
return self.async_show_form(
step_id="user",
description_placeholders={"api_url": API_URL},
data_schema=vol.Schema(
{
vol.Required(
CONF_API_TOKEN, default=user_input[CONF_API_TOKEN]
): str,
}
),
errors=self._errors,
)
async def async_step_site(self, user_input: dict[str, Any] = None):
"""Step to select site."""
self._errors = {}
assert self._sites is not None
api_token = self._api_token
if user_input is not None:
site_nmi = user_input[CONF_SITE_NMI]
sites = [site for site in self._sites if site.nmi == site_nmi]
site = sites[0]
site_id = site.id
name = user_input.get(CONF_SITE_NAME, site_id)
return self.async_create_entry(
title=name,
data={
CONF_SITE_ID: site_id,
CONF_API_TOKEN: api_token,
CONF_SITE_NMI: site.nmi,
},
)
user_input = {
CONF_API_TOKEN: api_token,
CONF_SITE_NMI: "",
CONF_SITE_NAME: "",
}
return self.async_show_form(
step_id="site",
data_schema=vol.Schema(
{
vol.Required(
CONF_SITE_NMI, default=user_input[CONF_SITE_NMI]
): vol.In([site.nmi for site in self._sites]),
vol.Optional(
CONF_SITE_NAME, default=user_input[CONF_SITE_NAME]
): str,
}
),
errors=self._errors,
)

11
homeassistant/components/amberelectric/const.py Normal file
View File

@ -0,0 +1,11 @@
"""Amber Electric Constants."""
import logging
DOMAIN = "amberelectric"
CONF_API_TOKEN = "api_token"
CONF_SITE_NAME = "site_name"
CONF_SITE_ID = "site_id"
CONF_SITE_NMI = "site_nmi"
LOGGER = logging.getLogger(__package__)
PLATFORMS = ["sensor"]

110
homeassistant/components/amberelectric/coordinator.py Normal file
View File

@ -0,0 +1,110 @@
"""Amber Electric Coordinator."""
from __future__ import annotations
from datetime import timedelta
from typing import Any
from amberelectric import ApiException
from amberelectric.api import amber_api
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 homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER
def is_current(interval: ActualInterval | CurrentInterval | ForecastInterval) -> bool:
"""Return true if the supplied interval is a CurrentInterval."""
return isinstance(interval, CurrentInterval)
def is_forecast(interval: ActualInterval | CurrentInterval | ForecastInterval) -> bool:
"""Return true if the supplied interval is a ForecastInterval."""
return isinstance(interval, ForecastInterval)
def is_general(interval: ActualInterval | CurrentInterval | ForecastInterval) -> bool:
"""Return true if the supplied interval is on the general channel."""
return interval.channel_type == ChannelType.GENERAL
def is_controlled_load(
interval: ActualInterval | CurrentInterval | ForecastInterval,
) -> bool:
"""Return true if the supplied interval is on the controlled load channel."""
return interval.channel_type == ChannelType.CONTROLLED_LOAD
def is_feed_in(interval: ActualInterval | CurrentInterval | ForecastInterval) -> bool:
"""Return true if the supplied interval is on the feed in channel."""
return interval.channel_type == ChannelType.FEED_IN
class AmberUpdateCoordinator(DataUpdateCoordinator):
"""AmberUpdateCoordinator - In charge of downloading the data for a site, which all the sensors read."""
def __init__(
self, hass: HomeAssistant, api: amber_api.AmberApi, site_id: str
) -> None:
"""Initialise the data service."""
super().__init__(
hass,
LOGGER,
name="amberelectric",
update_interval=timedelta(minutes=1),
)
self._api = api
self.site_id = site_id
def update_price_data(self) -> dict[str, dict[str, Any]]:
"""Update callback."""
result: dict[str, dict[str, Any]] = {
"current": {},
"forecasts": {},
"grid": {},
}
try:
data = self._api.get_current_price(self.site_id, next=48)
except ApiException as api_exception:
raise UpdateFailed("Missing price data, skipping update") from api_exception
current = [interval for interval in data if is_current(interval)]
forecasts = [interval for interval in data if is_forecast(interval)]
general = [interval for interval in current if is_general(interval)]
if len(general) == 0:
raise UpdateFailed("No general channel configured")
result["current"]["general"] = general[0]
result["forecasts"]["general"] = [
interval for interval in forecasts if is_general(interval)
]
result["grid"]["renewables"] = round(general[0].renewables)
controlled_load = [
interval for interval in current if is_controlled_load(interval)
]
if controlled_load:
result["current"]["controlled_load"] = controlled_load[0]
result["forecasts"]["controlled_load"] = [
interval for interval in forecasts if is_controlled_load(interval)
]
feed_in = [interval for interval in current if is_feed_in(interval)]
if feed_in:
result["current"]["feed_in"] = feed_in[0]
result["forecasts"]["feed_in"] = [
interval for interval in forecasts if is_feed_in(interval)
]
LOGGER.debug("Fetched new Amber data: %s", data)
return result
async def _async_update_data(self) -> dict[str, Any]:
"""Async update wrapper."""
return await self.hass.async_add_executor_job(self.update_price_data)

13
homeassistant/components/amberelectric/manifest.json Normal file
View File

@ -0,0 +1,13 @@
{
"domain": "amberelectric",
"name": "Amber Electric",
"documentation": "https://www.home-assistant.io/integrations/amberelectric",
"config_flow": true,
"codeowners": [
"@madpilot"
],
"requirements": [
"amberelectric==1.0.3"
],
"iot_class": "cloud_polling"
}

234
homeassistant/components/amberelectric/sensor.py Normal file
View File

@ -0,0 +1,234 @@
"""Amber Electric Sensor definitions."""
# There are three types of sensor: Current, Forecast and Grid
# Current and forecast will create general, controlled load and feed in as required
# At the moment renewables in the only grid sensor.
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from amberelectric.model.channel import ChannelType
from amberelectric.model.current_interval import CurrentInterval
from amberelectric.model.forecast_interval import ForecastInterval
from homeassistant.components.sensor import (
STATE_CLASS_MEASUREMENT,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION, CURRENCY_DOLLAR, ENERGY_KILO_WATT_HOUR
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import AmberUpdateCoordinator
ATTRIBUTION = "Data provided by Amber Electric"
ICONS = {
"general": "mdi:transmission-tower",
"controlled_load": "mdi:clock-outline",
"feed_in": "mdi:solar-power",
}
UNIT = f"{CURRENCY_DOLLAR}/{ENERGY_KILO_WATT_HOUR}"
def friendly_channel_type(channel_type: str) -> str:
"""Return a human readable version of the channel type."""
if channel_type == "controlled_load":
return "Controlled Load"
if channel_type == "feed_in":
return "Feed In"
return "General"
class AmberSensor(CoordinatorEntity, SensorEntity):
"""Amber Base Sensor."""
def __init__(
self,
coordinator: AmberUpdateCoordinator,
description: SensorEntityDescription,
channel_type: ChannelType,
) -> None:
"""Initialize the Sensor."""
super().__init__(coordinator)
self.site_id = coordinator.site_id
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}"
)
class AmberPriceSensor(AmberSensor):
"""Amber Price Sensor."""
@property
def native_value(self) -> str | None:
"""Return the current price in $/kWh."""
interval = self.coordinator.data[self.entity_description.key][self.channel_type]
if interval.channel_type == ChannelType.FEED_IN:
return round(interval.per_kwh, 0) / 100 * -1
return round(interval.per_kwh, 0) / 100
@property
def device_state_attributes(self) -> Mapping[str, Any] | None:
"""Return additional pieces of information about the price."""
interval = self.coordinator.data[self.entity_description.key][self.channel_type]
data: dict[str, Any] = {ATTR_ATTRIBUTION: ATTRIBUTION}
if interval is None:
return data
data["duration"] = interval.duration
data["date"] = interval.date.isoformat()
data["per_kwh"] = round(interval.per_kwh)
if interval.channel_type == ChannelType.FEED_IN:
data["per_kwh"] = data["per_kwh"] * -1
data["nem_date"] = interval.nem_time.isoformat()
data["spot_per_kwh"] = round(interval.spot_per_kwh)
data["start_time"] = interval.start_time.isoformat()
data["end_time"] = interval.end_time.isoformat()
data["renewables"] = round(interval.renewables)
data["estimate"] = interval.estimate
data["spike_status"] = interval.spike_status.value
data["channel_type"] = interval.channel_type.value
if interval.range is not None:
data["range_min"] = interval.range.min
data["range_max"] = interval.range.max
return data
class AmberForecastSensor(AmberSensor):
"""Amber Forecast Sensor."""
@property
def native_value(self) -> str | None:
"""Return the first forecast price in $/kWh."""
intervals = self.coordinator.data[self.entity_description.key][
self.channel_type
]
interval = intervals[0]
if interval.channel_type == ChannelType.FEED_IN:
return round(interval.per_kwh, 0) / 100 * -1
return round(interval.per_kwh, 0) / 100
@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][
self.channel_type
]
data = {
"forecasts": [],
"channel_type": intervals[0].channel_type.value,
ATTR_ATTRIBUTION: ATTRIBUTION,
}
for interval in intervals:
datum = {}
datum["duration"] = interval.duration
datum["date"] = interval.date.isoformat()
datum["nem_date"] = interval.nem_time.isoformat()
datum["per_kwh"] = round(interval.per_kwh)
if interval.channel_type == ChannelType.FEED_IN:
datum["per_kwh"] = datum["per_kwh"] * -1
datum["spot_per_kwh"] = round(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
if interval.range is not None:
datum["range_min"] = interval.range.min
datum["range_max"] = interval.range.max
data["forecasts"].append(datum)
return data
class AmberGridSensor(CoordinatorEntity, SensorEntity):
"""Sensor to show single grid specific values."""
def __init__(
self,
coordinator: AmberUpdateCoordinator,
description: SensorEntityDescription,
) -> None:
"""Initialize the Sensor."""
super().__init__(coordinator)
self.site_id = coordinator.site_id
self.entity_description = description
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."""
return self.coordinator.data["grid"][self.entity_description.key]
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]
current: dict[str, CurrentInterval] = coordinator.data["current"]
forecasts: dict[str, list[ForecastInterval]] = coordinator.data["forecasts"]
entities: list = []
for channel_type in current:
description = SensorEntityDescription(
key="current",
name=f"{entry.title} - {friendly_channel_type(channel_type)} Price",
native_unit_of_measurement=UNIT,
state_class=STATE_CLASS_MEASUREMENT,
icon=ICONS[channel_type],
)
entities.append(AmberPriceSensor(coordinator, description, channel_type))
for channel_type in forecasts:
description = SensorEntityDescription(
key="forecasts",
name=f"{entry.title} - {friendly_channel_type(channel_type)} Forecast",
native_unit_of_measurement=UNIT,
state_class=STATE_CLASS_MEASUREMENT,
icon=ICONS[channel_type],
)
entities.append(AmberForecastSensor(coordinator, description, channel_type))
renewables_description = SensorEntityDescription(
key="renewables",
name=f"{entry.title} - Renewables",
native_unit_of_measurement="%",
state_class=STATE_CLASS_MEASUREMENT,
icon="mdi:solar-power",
)
entities.append(AmberGridSensor(coordinator, renewables_description))
async_add_entities(entities)

22
homeassistant/components/amberelectric/strings.json Normal file
View File

@ -0,0 +1,22 @@
{
"config": {
"step": {
"user": {
"data": {
"api_token": "API Token",
"site_id": "Site ID"
},
"title": "Amber Electric",
"description": "Go to {api_url} to generate an API key"
},
"site": {
"data": {
"site_nmi": "Site NMI",
"site_name": "Site Name"
},
"title": "Amber Electric",
"description": "Select the NMI of the site you would like to add"
}
}
}
}

22
homeassistant/components/amberelectric/translations/en.json Normal file
View File

@ -0,0 +1,22 @@
{
"config": {
"step": {
"site": {
"data": {
"site_name": "Site Name",
"site_nmi": "Site NMI"
},
"description": "Select the NMI of the site you would like to add",
"title": "Amber Electric"
},
"user": {
"data": {
"api_token": "API Token",
"site_id": "Site ID"
},
"description": "Go to {api_url} to generate an API key",
"title": "Amber Electric"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -22,6 +22,7 @@ FLOWS = [
"alarmdecoder",
"almond",
"ambee",
"amberelectric",
"ambiclimate",
"ambient_station",
"apple_tv",

3
requirements_all.txt
View File

@ -281,6 +281,9 @@ alpha_vantage==2.3.1
# homeassistant.components.ambee
ambee==0.3.0
# homeassistant.components.amberelectric
amberelectric==1.0.3
# homeassistant.components.ambiclimate
ambiclimate==0.2.1

3
requirements_test_all.txt
View File

@ -199,6 +199,9 @@ airtouch4pyapi==1.0.5
# homeassistant.components.ambee
ambee==0.3.0
# homeassistant.components.amberelectric
amberelectric==1.0.3
# homeassistant.components.ambiclimate
ambiclimate==0.2.1

121
tests/components/amberelectric/helpers.py Normal file
View File

@ -0,0 +1,121 @@
"""Some common test functions for testing Amber components."""
from datetime import datetime, timedelta
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 dateutil import parser
def generate_actual_interval(
channel_type: ChannelType, end_time: datetime
) -> ActualInterval:
"""Generate a mock actual interval."""
start_time = end_time - timedelta(minutes=30)
return ActualInterval(
duration=30,
spot_per_kwh=1.0,
per_kwh=8.0,
date=start_time.date(),
nem_time=end_time,
start_time=start_time,
end_time=end_time,
renewables=50,
channel_type=channel_type.value,
spike_status=SpikeStatus.NO_SPIKE.value,
)
def generate_current_interval(
channel_type: ChannelType, end_time: datetime
) -> CurrentInterval:
"""Generate a mock current price."""
start_time = end_time - timedelta(minutes=30)
return CurrentInterval(
duration=30,
spot_per_kwh=1.0,
per_kwh=8.0,
date=start_time.date(),
nem_time=end_time,
start_time=start_time,
end_time=end_time,
renewables=50.6,
channel_type=channel_type.value,
spike_status=SpikeStatus.NO_SPIKE.value,
estimate=True,
)
def generate_forecast_interval(
channel_type: ChannelType, end_time: datetime
) -> ForecastInterval:
"""Generate a mock forecast interval."""
start_time = end_time - timedelta(minutes=30)
return ForecastInterval(
duration=30,
spot_per_kwh=1.1,
per_kwh=8.8,
date=start_time.date(),
nem_time=end_time,
start_time=start_time,
end_time=end_time,
renewables=50,
channel_type=channel_type.value,
spike_status=SpikeStatus.NO_SPIKE.value,
estimate=True,
)
GENERAL_ONLY_SITE_ID = "01FG2K6V5TB6X9W0EWPPMZD6MJ"
GENERAL_AND_CONTROLLED_SITE_ID = "01FG2MC8RF7GBC4KJXP3YFZ162"
GENERAL_AND_FEED_IN_SITE_ID = "01FG2MCD8KTRZR9MNNW84VP50S"
GENERAL_AND_CONTROLLED_FEED_IN_SITE_ID = "01FG2MCD8KTRZR9MNNW847S50S"
GENERAL_CHANNEL = [
generate_current_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T08:30:00+10:00")
),
generate_forecast_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T09:00:00+10:00")
),
generate_forecast_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T09:30:00+10:00")
),
generate_forecast_interval(
ChannelType.GENERAL, parser.parse("2021-09-21T10:00:00+10:00")
),
]
CONTROLLED_LOAD_CHANNEL = [
generate_current_interval(
ChannelType.CONTROLLED_LOAD, parser.parse("2021-09-21T08:30:00+10:00")
),
generate_forecast_interval(
ChannelType.CONTROLLED_LOAD, parser.parse("2021-09-21T09:00:00+10:00")
),
generate_forecast_interval(
ChannelType.CONTROLLED_LOAD, parser.parse("2021-09-21T09:30:00+10:00")
),
generate_forecast_interval(
ChannelType.CONTROLLED_LOAD, parser.parse("2021-09-21T10:00:00+10:00")
),
]
FEED_IN_CHANNEL = [
generate_current_interval(
ChannelType.FEED_IN, parser.parse("2021-09-21T08:30:00+10:00")
),
generate_forecast_interval(
ChannelType.FEED_IN, parser.parse("2021-09-21T09:00:00+10:00")
),
generate_forecast_interval(
ChannelType.FEED_IN, parser.parse("2021-09-21T09:30:00+10:00")
),
generate_forecast_interval(
ChannelType.FEED_IN, parser.parse("2021-09-21T10:00:00+10:00")
),
]

148
tests/components/amberelectric/test_config_flow.py Normal file
View File

@ -0,0 +1,148 @@
"""Tests for the Amber config flow."""
from typing import Generator
from unittest.mock import Mock, patch
from amberelectric import ApiException
from amberelectric.model.site import Site
import pytest
from homeassistant import data_entry_flow
from homeassistant.components.amberelectric.const import (
CONF_SITE_ID,
CONF_SITE_NAME,
CONF_SITE_NMI,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_TOKEN
from homeassistant.core import HomeAssistant
API_KEY = "psk_123456789"
@pytest.fixture(name="invalid_key_api")
def mock_invalid_key_api() -> Generator:
"""Return an authentication error."""
instance = Mock()
instance.get_sites.side_effect = ApiException(status=403)
with patch("amberelectric.api.AmberApi.create", return_value=instance):
yield instance
@pytest.fixture(name="api_error")
def mock_api_error() -> Generator:
"""Return an authentication error."""
instance = Mock()
instance.get_sites.side_effect = ApiException(status=500)
with patch("amberelectric.api.AmberApi.create", return_value=instance):
yield instance
@pytest.fixture(name="single_site_api")
def mock_single_site_api() -> Generator:
"""Return a single site."""
instance = Mock()
site = Site("01FG0AGP818PXK0DWHXJRRT2DH", "11111111111", [])
instance.get_sites.return_value = [site]
with patch("amberelectric.api.AmberApi.create", return_value=instance):
yield instance
@pytest.fixture(name="no_site_api")
def mock_no_site_api() -> Generator:
"""Return no site."""
instance = Mock()
instance.get_sites.return_value = []
with patch("amberelectric.api.AmberApi.create", return_value=instance):
yield instance
async def test_single_site(hass: HomeAssistant, single_site_api: Mock) -> None:
"""Test single site."""
initial_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert initial_result.get("type") == data_entry_flow.RESULT_TYPE_FORM
assert initial_result.get("step_id") == "user"
# Test filling in API key
enter_api_key_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_API_TOKEN: API_KEY},
)
assert enter_api_key_result.get("type") == data_entry_flow.RESULT_TYPE_FORM
assert enter_api_key_result.get("step_id") == "site"
select_site_result = await hass.config_entries.flow.async_configure(
enter_api_key_result["flow_id"],
{CONF_SITE_NMI: "11111111111", CONF_SITE_NAME: "Home"},
)
# Show available sites
assert select_site_result.get("type") == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert select_site_result.get("title") == "Home"
data = select_site_result.get("data")
assert data
assert data[CONF_API_TOKEN] == API_KEY
assert data[CONF_SITE_ID] == "01FG0AGP818PXK0DWHXJRRT2DH"
assert data[CONF_SITE_NMI] == "11111111111"
async def test_no_site(hass: HomeAssistant, no_site_api: Mock) -> None:
"""Test no site."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_API_TOKEN: "psk_123456789"},
)
assert result.get("type") == data_entry_flow.RESULT_TYPE_FORM
# Goes back to the user step
assert result.get("step_id") == "user"
assert result.get("errors") == {"api_token": "no_site"}
async def test_invalid_key(hass: HomeAssistant, invalid_key_api: Mock) -> None:
"""Test invalid api key."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") == data_entry_flow.RESULT_TYPE_FORM
assert result.get("step_id") == "user"
# Test filling in API key
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_API_TOKEN: "psk_123456789"},
)
assert result.get("type") == data_entry_flow.RESULT_TYPE_FORM
# Goes back to the user step
assert result.get("step_id") == "user"
assert result.get("errors") == {"api_token": "invalid_api_token"}
async def test_unknown_error(hass: HomeAssistant, api_error: Mock) -> None:
"""Test invalid api key."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") == data_entry_flow.RESULT_TYPE_FORM
assert result.get("step_id") == "user"
# Test filling in API key
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_API_TOKEN: "psk_123456789"},
)
assert result.get("type") == data_entry_flow.RESULT_TYPE_FORM
# Goes back to the user step
assert result.get("step_id") == "user"
assert result.get("errors") == {"api_token": "unknown_error"}

202
tests/components/amberelectric/test_coordinator.py Normal file
View File

@ -0,0 +1,202 @@
"""Tests for the Amber Electric Data Coordinator."""
from typing import Generator
from unittest.mock import Mock, patch
from amberelectric import ApiException
from amberelectric.model.channel import Channel, ChannelType
from amberelectric.model.site import Site
import pytest
from homeassistant.components.amberelectric.coordinator import AmberUpdateCoordinator
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import UpdateFailed
from tests.components.amberelectric.helpers import (
CONTROLLED_LOAD_CHANNEL,
FEED_IN_CHANNEL,
GENERAL_AND_CONTROLLED_SITE_ID,
GENERAL_AND_FEED_IN_SITE_ID,
GENERAL_CHANNEL,
GENERAL_ONLY_SITE_ID,
)
@pytest.fixture(name="current_price_api")
def mock_api_current_price() -> Generator:
"""Return an authentication error."""
instance = Mock()
general_site = Site(
GENERAL_ONLY_SITE_ID,
"11111111111",
[Channel(identifier="E1", type=ChannelType.GENERAL)],
)
general_and_controlled_load = Site(
GENERAL_AND_CONTROLLED_SITE_ID,
"11111111112",
[
Channel(identifier="E1", type=ChannelType.GENERAL),
Channel(identifier="E2", type=ChannelType.CONTROLLED_LOAD),
],
)
general_and_feed_in = Site(
GENERAL_AND_FEED_IN_SITE_ID,
"11111111113",
[
Channel(identifier="E1", type=ChannelType.GENERAL),
Channel(identifier="E2", type=ChannelType.FEED_IN),
],
)
instance.get_sites.return_value = [
general_site,
general_and_controlled_load,
general_and_feed_in,
]
with patch("amberelectric.api.AmberApi.create", return_value=instance):
yield instance
async def test_fetch_general_site(hass: HomeAssistant, current_price_api: Mock) -> None:
"""Test fetching a site with only a general channel."""
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()
current_price_api.get_current_price.assert_called_with(
GENERAL_ONLY_SITE_ID, next=48
)
assert result["current"].get("general") == GENERAL_CHANNEL[0]
assert result["forecasts"].get("general") == [
GENERAL_CHANNEL[1],
GENERAL_CHANNEL[2],
GENERAL_CHANNEL[3],
]
assert result["current"].get("controlled_load") is None
assert result["forecasts"].get("controlled_load") is None
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)
async def test_fetch_no_general_site(
hass: HomeAssistant, current_price_api: Mock
) -> None:
"""Test fetching a site with no general channel."""
current_price_api.get_current_price.return_value = CONTROLLED_LOAD_CHANNEL
data_service = AmberUpdateCoordinator(hass, current_price_api, GENERAL_ONLY_SITE_ID)
with pytest.raises(UpdateFailed):
await data_service._async_update_data()
current_price_api.get_current_price.assert_called_with(
GENERAL_ONLY_SITE_ID, next=48
)
async def test_fetch_api_error(hass: HomeAssistant, current_price_api: Mock) -> None:
"""Test that the old values are maintained if a second call fails."""
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()
current_price_api.get_current_price.assert_called_with(
GENERAL_ONLY_SITE_ID, next=48
)
assert result["current"].get("general") == GENERAL_CHANNEL[0]
assert result["forecasts"].get("general") == [
GENERAL_CHANNEL[1],
GENERAL_CHANNEL[2],
GENERAL_CHANNEL[3],
]
assert result["current"].get("controlled_load") is None
assert result["forecasts"].get("controlled_load") is None
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)
current_price_api.get_current_price.side_effect = ApiException(status=403)
with pytest.raises(UpdateFailed):
await data_service._async_update_data()
assert result["current"].get("general") == GENERAL_CHANNEL[0]
assert result["forecasts"].get("general") == [
GENERAL_CHANNEL[1],
GENERAL_CHANNEL[2],
GENERAL_CHANNEL[3],
]
assert result["current"].get("controlled_load") is None
assert result["forecasts"].get("controlled_load") is None
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)
async def test_fetch_general_and_controlled_load_site(
hass: HomeAssistant, current_price_api: Mock
) -> None:
"""Test fetching a site with a general and controlled load channel."""
current_price_api.get_current_price.return_value = (
GENERAL_CHANNEL + CONTROLLED_LOAD_CHANNEL
)
data_service = AmberUpdateCoordinator(
hass, current_price_api, GENERAL_AND_CONTROLLED_SITE_ID
)
result = await data_service._async_update_data()
current_price_api.get_current_price.assert_called_with(
GENERAL_AND_CONTROLLED_SITE_ID, next=48
)
assert result["current"].get("general") == GENERAL_CHANNEL[0]
assert result["forecasts"].get("general") == [
GENERAL_CHANNEL[1],
GENERAL_CHANNEL[2],
GENERAL_CHANNEL[3],
]
assert result["current"].get("controlled_load") is CONTROLLED_LOAD_CHANNEL[0]
assert result["forecasts"].get("controlled_load") == [
CONTROLLED_LOAD_CHANNEL[1],
CONTROLLED_LOAD_CHANNEL[2],
CONTROLLED_LOAD_CHANNEL[3],
]
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)
async def test_fetch_general_and_feed_in_site(
hass: HomeAssistant, current_price_api: Mock
) -> None:
"""Test fetching a site with a general and feed_in channel."""
current_price_api.get_current_price.return_value = GENERAL_CHANNEL + FEED_IN_CHANNEL
data_service = AmberUpdateCoordinator(
hass, current_price_api, GENERAL_AND_FEED_IN_SITE_ID
)
result = await data_service._async_update_data()
current_price_api.get_current_price.assert_called_with(
GENERAL_AND_FEED_IN_SITE_ID, next=48
)
assert result["current"].get("general") == GENERAL_CHANNEL[0]
assert result["forecasts"].get("general") == [
GENERAL_CHANNEL[1],
GENERAL_CHANNEL[2],
GENERAL_CHANNEL[3],
]
assert result["current"].get("controlled_load") is None
assert result["forecasts"].get("controlled_load") is None
assert result["current"].get("feed_in") is FEED_IN_CHANNEL[0]
assert result["forecasts"].get("feed_in") == [
FEED_IN_CHANNEL[1],
FEED_IN_CHANNEL[2],
FEED_IN_CHANNEL[3],
]
assert result["grid"]["renewables"] == round(GENERAL_CHANNEL[0].renewables)

282
tests/components/amberelectric/test_sensor.py Normal file
View File

@ -0,0 +1,282 @@
"""Test the Amber Electric Sensors."""
from typing import AsyncGenerator, List
from unittest.mock import Mock, patch
from amberelectric.model.current_interval import CurrentInterval
from amberelectric.model.range import Range
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 (
CONTROLLED_LOAD_CHANNEL,
FEED_IN_CHANNEL,
GENERAL_AND_CONTROLLED_SITE_ID,
GENERAL_AND_FEED_IN_SITE_ID,
GENERAL_CHANNEL,
GENERAL_ONLY_SITE_ID,
)
MOCK_API_TOKEN = "psk_0000000000000000"
@pytest.fixture
async def setup_general(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_general_and_controlled_load(hass) -> AsyncGenerator:
"""Set up general channel and controller load channel."""
MockConfigEntry(
domain="amberelectric",
data={
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_SITE_ID: GENERAL_AND_CONTROLLED_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 + CONTROLLED_LOAD_CHANNEL
)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
yield mock_update.return_value
@pytest.fixture
async def setup_general_and_feed_in(hass) -> AsyncGenerator:
"""Set up general channel and feed in channel."""
MockConfigEntry(
domain="amberelectric",
data={
CONF_API_TOKEN: MOCK_API_TOKEN,
CONF_SITE_ID: GENERAL_AND_FEED_IN_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 + FEED_IN_CHANNEL
)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
yield mock_update.return_value
async def test_general_price_sensor(hass: HomeAssistant, setup_general: Mock) -> None:
"""Test the General Price sensor."""
assert len(hass.states.async_all()) == 3
price = hass.states.get("sensor.mock_title_general_price")
assert price
assert price.state == "0.08"
attributes = price.attributes
assert attributes["duration"] == 30
assert attributes["date"] == "2021-09-21"
assert attributes["per_kwh"] == 8
assert attributes["nem_date"] == "2021-09-21T08:30:00+10:00"
assert attributes["spot_per_kwh"] == 1
assert attributes["start_time"] == "2021-09-21T08:00:00+10:00"
assert attributes["end_time"] == "2021-09-21T08:30:00+10:00"
assert attributes["renewables"] == 51
assert attributes["estimate"] is True
assert attributes["spike_status"] == "none"
assert attributes["channel_type"] == "general"
assert attributes["attribution"] == "Data provided by Amber Electric"
assert attributes.get("range_min") is None
assert attributes.get("range_max") is None
with_range: List[CurrentInterval] = GENERAL_CHANNEL
with_range[0].range = Range(7.8, 12.4)
setup_general.get_current_price.return_value = with_range
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
price = hass.states.get("sensor.mock_title_general_price")
assert price
attributes = price.attributes
assert attributes.get("range_min") == 7.8
assert attributes.get("range_max") == 12.4
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
print(hass.states)
price = hass.states.get("sensor.mock_title_controlled_load_price")
assert price
assert price.state == "0.08"
attributes = price.attributes
assert attributes["duration"] == 30
assert attributes["date"] == "2021-09-21"
assert attributes["per_kwh"] == 8
assert attributes["nem_date"] == "2021-09-21T08:30:00+10:00"
assert attributes["spot_per_kwh"] == 1
assert attributes["start_time"] == "2021-09-21T08:00:00+10:00"
assert attributes["end_time"] == "2021-09-21T08:30:00+10:00"
assert attributes["renewables"] == 51
assert attributes["estimate"] is True
assert attributes["spike_status"] == "none"
assert attributes["channel_type"] == "controlledLoad"
assert attributes["attribution"] == "Data provided by Amber Electric"
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
print(hass.states)
price = hass.states.get("sensor.mock_title_feed_in_price")
assert price
assert price.state == "-0.08"
attributes = price.attributes
assert attributes["duration"] == 30
assert attributes["date"] == "2021-09-21"
assert attributes["per_kwh"] == -8
assert attributes["nem_date"] == "2021-09-21T08:30:00+10:00"
assert attributes["spot_per_kwh"] == 1
assert attributes["start_time"] == "2021-09-21T08:00:00+10:00"
assert attributes["end_time"] == "2021-09-21T08:30:00+10:00"
assert attributes["renewables"] == 51
assert attributes["estimate"] is True
assert attributes["spike_status"] == "none"
assert attributes["channel_type"] == "feedIn"
assert attributes["attribution"] == "Data provided by Amber Electric"
async def test_general_forecast_sensor(
hass: HomeAssistant, setup_general: Mock
) -> None:
"""Test the General Forecast sensor."""
assert len(hass.states.async_all()) == 3
price = hass.states.get("sensor.mock_title_general_forecast")
assert price
assert price.state == "0.09"
attributes = price.attributes
assert attributes["channel_type"] == "general"
assert attributes["attribution"] == "Data provided by Amber Electric"
first_forecast = attributes["forecasts"][0]
assert first_forecast["duration"] == 30
assert first_forecast["date"] == "2021-09-21"
assert first_forecast["per_kwh"] == 9
assert first_forecast["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first_forecast["spot_per_kwh"] == 1
assert first_forecast["start_time"] == "2021-09-21T08:30:00+10:00"
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.get("range_min") is None
assert first_forecast.get("range_max") is None
with_range: List[CurrentInterval] = GENERAL_CHANNEL
with_range[1].range = Range(7.8, 12.4)
setup_general.get_current_price.return_value = with_range
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
price = hass.states.get("sensor.mock_title_general_forecast")
assert price
attributes = price.attributes
first_forecast = attributes["forecasts"][0]
assert first_forecast.get("range_min") == 7.8
assert first_forecast.get("range_max") == 12.4
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
price = hass.states.get("sensor.mock_title_controlled_load_forecast")
assert price
assert price.state == "0.09"
attributes = price.attributes
assert attributes["channel_type"] == "controlledLoad"
assert attributes["attribution"] == "Data provided by Amber Electric"
first_forecast = attributes["forecasts"][0]
assert first_forecast["duration"] == 30
assert first_forecast["date"] == "2021-09-21"
assert first_forecast["per_kwh"] == 9
assert first_forecast["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first_forecast["spot_per_kwh"] == 1
assert first_forecast["start_time"] == "2021-09-21T08:30:00+10:00"
assert first_forecast["end_time"] == "2021-09-21T09:00:00+10:00"
assert first_forecast["renewables"] == 50
assert first_forecast["spike_status"] == "none"
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
price = hass.states.get("sensor.mock_title_feed_in_forecast")
assert price
assert price.state == "-0.09"
attributes = price.attributes
assert attributes["channel_type"] == "feedIn"
assert attributes["attribution"] == "Data provided by Amber Electric"
first_forecast = attributes["forecasts"][0]
assert first_forecast["duration"] == 30
assert first_forecast["date"] == "2021-09-21"
assert first_forecast["per_kwh"] == -9
assert first_forecast["nem_date"] == "2021-09-21T09:00:00+10:00"
assert first_forecast["spot_per_kwh"] == 1
assert first_forecast["start_time"] == "2021-09-21T08:30:00+10:00"
assert first_forecast["end_time"] == "2021-09-21T09:00:00+10:00"
assert first_forecast["renewables"] == 50
assert first_forecast["spike_status"] == "none"
def test_renewable_sensor(hass: HomeAssistant, setup_general) -> None:
"""Testing the creation of the Amber renewables sensor."""
assert len(hass.states.async_all()) == 3
sensor = hass.states.get("sensor.mock_title_renewables")
assert sensor
assert sensor.state == "51"