jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add price service call to Tibber (#117366)

Browse Source 
Co-authored-by: Daniel Hjelseth Høyer <mail@dahoiv.net>
Co-authored-by: Jason R. Coombs <jaraco@jaraco.com>
This commit is contained in:
Bas Brussee 2024-06-10 21:16:51 +02:00 committed by GitHub
parent 6184fd26d3
commit b30a924e03
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 397 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
homeassistant/components/tibber/__init__.py
View File

@ -21,6 +21,7 @@ from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util
from .const import DATA_HASS_CONFIG, DOMAIN
from .services import async_setup_services
PLATFORMS = [Platform.NOTIFY, Platform.SENSOR]
@ -33,6 +34,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Tibber component."""
hass.data[DATA_HASS_CONFIG] = config
async_setup_services(hass)
return True

5
homeassistant/components/tibber/icons.json Normal file
View File

@ -0,0 +1,5 @@
{
"services": {
"get_prices": "mdi:cash"
}
}

106
homeassistant/components/tibber/services.py Normal file
View File

@ -0,0 +1,106 @@
"""Services for Tibber integration."""
from __future__ import annotations
import datetime as dt
from datetime import date, datetime
from functools import partial
from typing import Any, Final
import voluptuous as vol
from homeassistant.core import (
HomeAssistant,
ServiceCall,
ServiceResponse,
SupportsResponse,
callback,
)
from homeassistant.exceptions import ServiceValidationError
from homeassistant.util import dt as dt_util
from .const import DOMAIN
PRICE_SERVICE_NAME = "get_prices"
ATTR_START: Final = "start"
ATTR_END: Final = "end"
SERVICE_SCHEMA: Final = vol.Schema(
{
vol.Optional(ATTR_START): str,
vol.Optional(ATTR_END): str,
}
)
async def __get_prices(call: ServiceCall, *, hass: HomeAssistant) -> ServiceResponse:
tibber_connection = hass.data[DOMAIN]
start = __get_date(call.data.get(ATTR_START), "start")
end = __get_date(call.data.get(ATTR_END), "end")
if start >= end:
end = start + dt.timedelta(days=1)
tibber_prices: dict[str, Any] = {}
for tibber_home in tibber_connection.get_homes(only_active=True):
home_nickname = tibber_home.name
price_info = tibber_home.info["viewer"]["home"]["currentSubscription"][
"priceInfo"
]
price_data = [
{
"start_time": dt.datetime.fromisoformat(price["startsAt"]),
"price": price["total"],
"level": price["level"],
}
for key in ("today", "tomorrow")
for price in price_info[key]
]
selected_data = [
price
for price in price_data
if price["start_time"].replace(tzinfo=None) >= start
and price["start_time"].replace(tzinfo=None) < end
]
tibber_prices[home_nickname] = selected_data
return {"prices": tibber_prices}
def __get_date(date_input: str | None, mode: str | None) -> date | datetime:
"""Get date."""
if not date_input:
if mode == "end":
increment = dt.timedelta(days=1)
else:
increment = dt.timedelta()
return datetime.fromisoformat(dt_util.now().date().isoformat()) + increment
if value := dt_util.parse_datetime(date_input):
return value
raise ServiceValidationError(
"Invalid datetime provided.",
translation_domain=DOMAIN,
translation_key="invalid_date",
translation_placeholders={
"date": date_input,
},
)
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up services for Tibber integration."""
hass.services.async_register(
DOMAIN,
PRICE_SERVICE_NAME,
partial(__get_prices, hass=hass),
schema=SERVICE_SCHEMA,
supports_response=SupportsResponse.ONLY,
)

12
homeassistant/components/tibber/services.yaml Normal file
View File

@ -0,0 +1,12 @@
get_prices:
fields:
start:
required: false
example: "2024-01-01 00:00:00"
selector:
datetime:
end:
required: false
example: "2024-01-01 23:00:00"
selector:
datetime:

16
homeassistant/components/tibber/strings.json
View File

@ -84,6 +84,22 @@
}
}
},
"services": {
"get_prices": {
"name": "Get enegry prices",
"description": "Get hourly energy prices from Tibber",
"fields": {
"start": {
"name": "Start",
"description": "Specifies the date and time from which to retrieve prices. Defaults to today if omitted."
},
"end": {
"name": "End",
"description": "Specifies the date and time until which to retrieve prices. Defaults to the last hour of today if omitted."
}
}
}
},
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"

254
tests/components/tibber/test_services.py Normal file
View File

@ -0,0 +1,254 @@
"""Test service for Tibber integration."""
import asyncio
import datetime as dt
from unittest.mock import MagicMock
import pytest
from homeassistant.components.tibber.const import DOMAIN
from homeassistant.components.tibber.services import PRICE_SERVICE_NAME, __get_prices
from homeassistant.core import ServiceCall
from homeassistant.exceptions import ServiceValidationError
def generate_mock_home_data():
"""Create mock data from the tibber connection."""
today = remove_microseconds(dt.datetime.now())
tomorrow = remove_microseconds(today + dt.timedelta(days=1))
mock_homes = [
MagicMock(
name="first_home",
info={
"viewer": {
"home": {
"currentSubscription": {
"priceInfo": {
"today": [
{
"startsAt": today.isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"startsAt": (
today + dt.timedelta(hours=1)
).isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
"tomorrow": [
{
"startsAt": tomorrow.isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"startsAt": (
tomorrow + dt.timedelta(hours=1)
).isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
}
}
}
}
},
),
MagicMock(
name="second_home",
info={
"viewer": {
"home": {
"currentSubscription": {
"priceInfo": {
"today": [
{
"startsAt": today.isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"startsAt": (
today + dt.timedelta(hours=1)
).isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
"tomorrow": [
{
"startsAt": tomorrow.isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"startsAt": (
tomorrow + dt.timedelta(hours=1)
).isoformat(),
"total": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
}
}
}
}
},
),
]
mock_homes[0].name = "first_home"
mock_homes[1].name = "second_home"
return mock_homes
def create_mock_tibber_connection():
"""Create a mock tibber connection."""
tibber_connection = MagicMock()
tibber_connection.get_homes.return_value = generate_mock_home_data()
return tibber_connection
def create_mock_hass():
"""Create a mock hass object."""
mock_hass = MagicMock
mock_hass.data = {"tibber": create_mock_tibber_connection()}
return mock_hass
def remove_microseconds(dt):
"""Remove microseconds from a datetime object."""
return dt.replace(microsecond=0)
async def test_get_prices():
"""Test __get_prices with mock data."""
today = remove_microseconds(dt.datetime.now())
tomorrow = remove_microseconds(dt.datetime.now() + dt.timedelta(days=1))
call = ServiceCall(
DOMAIN,
PRICE_SERVICE_NAME,
{"start": today.date().isoformat(), "end": tomorrow.date().isoformat()},
)
result = await __get_prices(call, hass=create_mock_hass())
assert result == {
"prices": {
"first_home": [
{
"start_time": today,
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"start_time": today + dt.timedelta(hours=1),
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
"second_home": [
{
"start_time": today,
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"start_time": today + dt.timedelta(hours=1),
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
}
}
async def test_get_prices_no_input():
"""Test __get_prices with no input."""
today = remove_microseconds(dt.datetime.now())
call = ServiceCall(DOMAIN, PRICE_SERVICE_NAME, {})
result = await __get_prices(call, hass=create_mock_hass())
assert result == {
"prices": {
"first_home": [
{
"start_time": today,
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"start_time": today + dt.timedelta(hours=1),
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
"second_home": [
{
"start_time": today,
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"start_time": today + dt.timedelta(hours=1),
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
}
}
async def test_get_prices_start_tomorrow():
"""Test __get_prices with start date tomorrow."""
tomorrow = remove_microseconds(dt.datetime.now() + dt.timedelta(days=1))
call = ServiceCall(
DOMAIN, PRICE_SERVICE_NAME, {"start": tomorrow.date().isoformat()}
)
result = await __get_prices(call, hass=create_mock_hass())
assert result == {
"prices": {
"first_home": [
{
"start_time": tomorrow,
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"start_time": tomorrow + dt.timedelta(hours=1),
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
"second_home": [
{
"start_time": tomorrow,
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
{
"start_time": tomorrow + dt.timedelta(hours=1),
"price": 0.46914,
"level": "VERY_EXPENSIVE",
},
],
}
}
async def test_get_prices_invalid_input():
"""Test __get_prices with invalid input."""
call = ServiceCall(DOMAIN, PRICE_SERVICE_NAME, {"start": "test"})
task = asyncio.create_task(__get_prices(call, hass=create_mock_hass()))
with pytest.raises(ServiceValidationError) as excinfo:
await task
assert "Invalid datetime provided." in str(excinfo.value)