Add config flow to Tankerkoenig (#68386)
Michael
GitHub
parent
f5a13fc51b
commit
2f7aeb64d2
|
|
@ -1195,7 +1195,9 @@ omit =
|
|||
homeassistant/components/tado/sensor.py
|
||||
homeassistant/components/tado/water_heater.py
|
||||
homeassistant/components/tank_utility/sensor.py
|
||||
homeassistant/components/tankerkoenig/*
|
||||
homeassistant/components/tankerkoenig/__init__.py
|
||||
homeassistant/components/tankerkoenig/const.py
|
||||
homeassistant/components/tankerkoenig/sensor.py
|
||||
homeassistant/components/tapsaff/binary_sensor.py
|
||||
homeassistant/components/tautulli/const.py
|
||||
homeassistant/components/tautulli/coordinator.py
|
||||
|
|
|
|||
|
|
@ -1009,7 +1009,8 @@ build.json @home-assistant/supervisor
|
|||
/tests/components/tag/ @balloob @dmulcahey
|
||||
/homeassistant/components/tailscale/ @frenck
|
||||
/tests/components/tailscale/ @frenck
|
||||
/homeassistant/components/tankerkoenig/ @guillempages
|
||||
/homeassistant/components/tankerkoenig/ @guillempages @mib1185
|
||||
/tests/components/tankerkoenig/ @guillempages @mib1185
|
||||
/homeassistant/components/tapsaff/ @bazwilliams
|
||||
/homeassistant/components/tasmota/ @emontnemery
|
||||
/tests/components/tasmota/ @emontnemery
|
||||
|
|
|
|||
|
|
@ -1,67 +1,82 @@
|
|||
"""Ask tankerkoenig.de for petrol price information."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from math import ceil
|
||||
|
||||
import pytankerkoenig
|
||||
from requests.exceptions import RequestException
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_API_KEY,
|
||||
CONF_LATITUDE,
|
||||
CONF_LOCATION,
|
||||
CONF_LONGITUDE,
|
||||
CONF_NAME,
|
||||
CONF_RADIUS,
|
||||
CONF_SCAN_INTERVAL,
|
||||
CONF_SHOW_ON_MAP,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.discovery import async_load_platform
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import CONF_FUEL_TYPES, CONF_STATIONS, DOMAIN, FUEL_TYPES
|
||||
from .const import (
|
||||
CONF_FUEL_TYPES,
|
||||
CONF_STATIONS,
|
||||
DEFAULT_RADIUS,
|
||||
DEFAULT_SCAN_INTERVAL,
|
||||
DOMAIN,
|
||||
FUEL_TYPES,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_RADIUS = 2
|
||||
DEFAULT_SCAN_INTERVAL = timedelta(minutes=30)
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_API_KEY): cv.string,
|
||||
vol.Optional(
|
||||
CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
|
||||
): cv.time_period,
|
||||
vol.Optional(CONF_FUEL_TYPES, default=FUEL_TYPES): vol.All(
|
||||
cv.ensure_list, [vol.In(FUEL_TYPES)]
|
||||
),
|
||||
vol.Inclusive(
|
||||
CONF_LATITUDE,
|
||||
"coordinates",
|
||||
"Latitude and longitude must exist together",
|
||||
): cv.latitude,
|
||||
vol.Inclusive(
|
||||
CONF_LONGITUDE,
|
||||
"coordinates",
|
||||
"Latitude and longitude must exist together",
|
||||
): cv.longitude,
|
||||
vol.Optional(CONF_RADIUS, default=DEFAULT_RADIUS): vol.All(
|
||||
cv.positive_int, vol.Range(min=1)
|
||||
),
|
||||
vol.Optional(CONF_STATIONS, default=[]): vol.All(
|
||||
cv.ensure_list, [cv.string]
|
||||
),
|
||||
vol.Optional(CONF_SHOW_ON_MAP, default=True): cv.boolean,
|
||||
}
|
||||
)
|
||||
},
|
||||
vol.All(
|
||||
cv.deprecated(DOMAIN),
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_API_KEY): cv.string,
|
||||
vol.Optional(
|
||||
CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
|
||||
): cv.time_period,
|
||||
vol.Optional(CONF_FUEL_TYPES, default=FUEL_TYPES): vol.All(
|
||||
cv.ensure_list, [vol.In(FUEL_TYPES)]
|
||||
),
|
||||
vol.Inclusive(
|
||||
CONF_LATITUDE,
|
||||
"coordinates",
|
||||
"Latitude and longitude must exist together",
|
||||
): cv.latitude,
|
||||
vol.Inclusive(
|
||||
CONF_LONGITUDE,
|
||||
"coordinates",
|
||||
"Latitude and longitude must exist together",
|
||||
): cv.longitude,
|
||||
vol.Optional(CONF_RADIUS, default=DEFAULT_RADIUS): vol.All(
|
||||
cv.positive_int, vol.Range(min=1)
|
||||
),
|
||||
vol.Optional(CONF_STATIONS, default=[]): vol.All(
|
||||
cv.ensure_list, [cv.string]
|
||||
),
|
||||
vol.Optional(CONF_SHOW_ON_MAP, default=True): cv.boolean,
|
||||
}
|
||||
)
|
||||
},
|
||||
),
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set the tankerkoenig component up."""
|
||||
|
|
@ -69,106 +84,119 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|||
return True
|
||||
|
||||
conf = config[DOMAIN]
|
||||
|
||||
_LOGGER.debug("Setting up integration")
|
||||
|
||||
tankerkoenig = TankerkoenigData(hass, conf)
|
||||
|
||||
latitude = conf.get(CONF_LATITUDE, hass.config.latitude)
|
||||
longitude = conf.get(CONF_LONGITUDE, hass.config.longitude)
|
||||
radius = conf[CONF_RADIUS]
|
||||
additional_stations = conf[CONF_STATIONS]
|
||||
|
||||
setup_ok = await hass.async_add_executor_job(
|
||||
tankerkoenig.setup, latitude, longitude, radius, additional_stations
|
||||
)
|
||||
if not setup_ok:
|
||||
_LOGGER.error("Could not setup integration")
|
||||
return False
|
||||
|
||||
hass.data[DOMAIN] = tankerkoenig
|
||||
|
||||
hass.async_create_task(
|
||||
async_load_platform(
|
||||
hass,
|
||||
SENSOR_DOMAIN,
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
discovered=tankerkoenig.stations,
|
||||
hass_config=conf,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={
|
||||
CONF_NAME: "Home",
|
||||
CONF_API_KEY: conf[CONF_API_KEY],
|
||||
CONF_FUEL_TYPES: conf[CONF_FUEL_TYPES],
|
||||
CONF_LOCATION: {
|
||||
"latitude": conf.get(CONF_LATITUDE, hass.config.latitude),
|
||||
"longitude": conf.get(CONF_LONGITUDE, hass.config.longitude),
|
||||
},
|
||||
CONF_RADIUS: conf[CONF_RADIUS],
|
||||
CONF_STATIONS: conf[CONF_STATIONS],
|
||||
CONF_SHOW_ON_MAP: conf[CONF_SHOW_ON_MAP],
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class TankerkoenigData:
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set a tankerkoenig configuration entry up."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
hass.data[DOMAIN][
|
||||
entry.unique_id
|
||||
] = coordinator = TankerkoenigDataUpdateCoordinator(
|
||||
hass,
|
||||
entry,
|
||||
_LOGGER,
|
||||
name=entry.unique_id or DOMAIN,
|
||||
update_interval=DEFAULT_SCAN_INTERVAL,
|
||||
)
|
||||
|
||||
try:
|
||||
setup_ok = await hass.async_add_executor_job(coordinator.setup)
|
||||
except RequestException as err:
|
||||
raise ConfigEntryNotReady from err
|
||||
if not setup_ok:
|
||||
_LOGGER.error("Could not setup integration")
|
||||
return False
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
|
||||
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload Tankerkoenig config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.unique_id)
|
||||
|
||||
return unload_ok
|
||||
|
||||
|
||||
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
||||
|
||||
class TankerkoenigDataUpdateCoordinator(DataUpdateCoordinator):
|
||||
"""Get the latest data from the API."""
|
||||
|
||||
def __init__(self, hass, conf):
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
logger: logging.Logger,
|
||||
name: str,
|
||||
update_interval: int,
|
||||
) -> None:
|
||||
"""Initialize the data object."""
|
||||
self._api_key = conf[CONF_API_KEY]
|
||||
self.stations = {}
|
||||
self.fuel_types = conf[CONF_FUEL_TYPES]
|
||||
self.update_interval = conf[CONF_SCAN_INTERVAL]
|
||||
self.show_on_map = conf[CONF_SHOW_ON_MAP]
|
||||
|
||||
super().__init__(
|
||||
hass=hass,
|
||||
logger=logger,
|
||||
name=name,
|
||||
update_interval=timedelta(minutes=update_interval),
|
||||
)
|
||||
|
||||
self._api_key = entry.data[CONF_API_KEY]
|
||||
self._selected_stations = entry.data[CONF_STATIONS]
|
||||
self._hass = hass
|
||||
self.stations: dict[str, dict] = {}
|
||||
self.fuel_types = entry.data[CONF_FUEL_TYPES]
|
||||
self.show_on_map = entry.options[CONF_SHOW_ON_MAP]
|
||||
|
||||
def setup(self, latitude, longitude, radius, additional_stations):
|
||||
"""Set up the tankerkoenig API.
|
||||
|
||||
Read the initial data from the server, to initialize the list of fuel stations to monitor.
|
||||
"""
|
||||
_LOGGER.debug("Fetching data for (%s, %s) rad: %s", latitude, longitude, radius)
|
||||
try:
|
||||
data = pytankerkoenig.getNearbyStations(
|
||||
self._api_key, latitude, longitude, radius, "all", "dist"
|
||||
)
|
||||
except pytankerkoenig.customException as err:
|
||||
data = {"ok": False, "message": err, "exception": True}
|
||||
_LOGGER.debug("Received data: %s", data)
|
||||
if not data["ok"]:
|
||||
_LOGGER.error(
|
||||
"Setup for sensors was unsuccessful. Error occurred while fetching data from tankerkoenig.de: %s",
|
||||
data["message"],
|
||||
)
|
||||
return False
|
||||
|
||||
# Add stations found via location + radius
|
||||
if not (nearby_stations := data["stations"]):
|
||||
if not additional_stations:
|
||||
_LOGGER.error(
|
||||
"Could not find any station in range."
|
||||
"Try with a bigger radius or manually specify stations in additional_stations"
|
||||
)
|
||||
return False
|
||||
_LOGGER.warning(
|
||||
"Could not find any station in range. Will only use manually specified stations"
|
||||
)
|
||||
else:
|
||||
for station in nearby_stations:
|
||||
self.add_station(station)
|
||||
|
||||
# Add manually specified additional stations
|
||||
for station_id in additional_stations:
|
||||
def setup(self):
|
||||
"""Set up the tankerkoenig API."""
|
||||
for station_id in self._selected_stations:
|
||||
try:
|
||||
additional_station_data = pytankerkoenig.getStationData(
|
||||
self._api_key, station_id
|
||||
)
|
||||
station_data = pytankerkoenig.getStationData(self._api_key, station_id)
|
||||
except pytankerkoenig.customException as err:
|
||||
additional_station_data = {
|
||||
station_data = {
|
||||
"ok": False,
|
||||
"message": err,
|
||||
"exception": True,
|
||||
}
|
||||
|
||||
if not additional_station_data["ok"]:
|
||||
if not station_data["ok"]:
|
||||
_LOGGER.error(
|
||||
"Error when adding station %s:\n %s",
|
||||
station_id,
|
||||
additional_station_data["message"],
|
||||
station_data["message"],
|
||||
)
|
||||
return False
|
||||
self.add_station(additional_station_data["station"])
|
||||
self.add_station(station_data["station"])
|
||||
if len(self.stations) > 10:
|
||||
_LOGGER.warning(
|
||||
"Found more than 10 stations to check. "
|
||||
|
|
@ -177,7 +205,7 @@ class TankerkoenigData:
|
|||
)
|
||||
return True
|
||||
|
||||
async def fetch_data(self):
|
||||
async def _async_update_data(self):
|
||||
"""Get the latest data from tankerkoenig.de."""
|
||||
_LOGGER.debug("Fetching new data from tankerkoenig.de")
|
||||
station_ids = list(self.stations)
|
||||
|
|
@ -198,10 +226,10 @@ class TankerkoenigData:
|
|||
_LOGGER.error(
|
||||
"Error fetching data from tankerkoenig.de: %s", data["message"]
|
||||
)
|
||||
raise TankerkoenigError(data["message"])
|
||||
raise UpdateFailed(data["message"])
|
||||
if "prices" not in data:
|
||||
_LOGGER.error("Did not receive price information from tankerkoenig.de")
|
||||
raise TankerkoenigError("No prices in data")
|
||||
raise UpdateFailed("No prices in data")
|
||||
prices.update(data["prices"])
|
||||
return prices
|
||||
|
||||
|
|
@ -216,7 +244,3 @@ class TankerkoenigData:
|
|||
|
||||
self.stations[station_id] = station
|
||||
_LOGGER.debug("add_station called for station: %s", station)
|
||||
|
||||
|
||||
class TankerkoenigError(HomeAssistantError):
|
||||
"""An error occurred while contacting tankerkoenig.de."""
|
||||
|
|
|
|||
|
|
@ -0,0 +1,224 @@
|
|||
"""Config flow for Tankerkoenig."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pytankerkoenig import customException, getNearbyStations
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import (
|
||||
CONF_API_KEY,
|
||||
CONF_LATITUDE,
|
||||
CONF_LOCATION,
|
||||
CONF_LONGITUDE,
|
||||
CONF_NAME,
|
||||
CONF_RADIUS,
|
||||
CONF_SHOW_ON_MAP,
|
||||
CONF_UNIT_OF_MEASUREMENT,
|
||||
LENGTH_KILOMETERS,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.selector import selector
|
||||
|
||||
from .const import CONF_FUEL_TYPES, CONF_STATIONS, DEFAULT_RADIUS, DOMAIN, FUEL_TYPES
|
||||
|
||||
|
||||
class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Init the FlowHandler."""
|
||||
super().__init__()
|
||||
self._data: dict[str, Any] = {}
|
||||
self._stations: dict[str, str] = {}
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(
|
||||
config_entry: config_entries.ConfigEntry,
|
||||
) -> OptionsFlowHandler:
|
||||
"""Get the options flow for this handler."""
|
||||
return OptionsFlowHandler(config_entry)
|
||||
|
||||
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
|
||||
"""Import YAML configuration."""
|
||||
await self.async_set_unique_id(
|
||||
f"{config[CONF_LOCATION][CONF_LATITUDE]}_{config[CONF_LOCATION][CONF_LONGITUDE]}"
|
||||
)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
selected_station_ids: list[str] = []
|
||||
# add all nearby stations
|
||||
nearby_stations = await self._get_nearby_stations(config)
|
||||
for station in nearby_stations.get("stations", []):
|
||||
selected_station_ids.append(station["id"])
|
||||
|
||||
# add all manual added stations
|
||||
for station_id in config[CONF_STATIONS]:
|
||||
selected_station_ids.append(station_id)
|
||||
|
||||
return self._create_entry(
|
||||
data={
|
||||
CONF_NAME: "Home",
|
||||
CONF_API_KEY: config[CONF_API_KEY],
|
||||
CONF_FUEL_TYPES: config[CONF_FUEL_TYPES],
|
||||
CONF_LOCATION: config[CONF_LOCATION],
|
||||
CONF_RADIUS: config[CONF_RADIUS],
|
||||
CONF_STATIONS: selected_station_ids,
|
||||
},
|
||||
options={
|
||||
CONF_SHOW_ON_MAP: config[CONF_SHOW_ON_MAP],
|
||||
},
|
||||
)
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
if not user_input:
|
||||
return self._show_form_user()
|
||||
|
||||
await self.async_set_unique_id(
|
||||
f"{user_input[CONF_LOCATION][CONF_LATITUDE]}_{user_input[CONF_LOCATION][CONF_LONGITUDE]}"
|
||||
)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
data = await self._get_nearby_stations(user_input)
|
||||
if not data.get("ok"):
|
||||
return self._show_form_user(
|
||||
user_input, errors={CONF_API_KEY: "invalid_auth"}
|
||||
)
|
||||
if stations := data.get("stations"):
|
||||
for station in stations:
|
||||
self._stations[
|
||||
station["id"]
|
||||
] = f"{station['brand']} {station['street']} {station['houseNumber']} - ({station['dist']}km)"
|
||||
|
||||
else:
|
||||
return self._show_form_user(user_input, errors={CONF_RADIUS: "no_stations"})
|
||||
|
||||
self._data = user_input
|
||||
|
||||
return await self.async_step_select_station()
|
||||
|
||||
async def async_step_select_station(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the step select_station of a flow initialized by the user."""
|
||||
if not user_input:
|
||||
return self.async_show_form(
|
||||
step_id="select_station",
|
||||
description_placeholders={"stations_count": len(self._stations)},
|
||||
data_schema=vol.Schema(
|
||||
{vol.Required(CONF_STATIONS): cv.multi_select(self._stations)}
|
||||
),
|
||||
)
|
||||
|
||||
return self._create_entry(
|
||||
data={**self._data, **user_input},
|
||||
options={CONF_SHOW_ON_MAP: True},
|
||||
)
|
||||
|
||||
def _show_form_user(
|
||||
self,
|
||||
user_input: dict[str, Any] | None = None,
|
||||
errors: dict[str, Any] | None = None,
|
||||
) -> FlowResult:
|
||||
if user_input is None:
|
||||
user_input = {}
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
CONF_NAME, default=user_input.get(CONF_NAME, "")
|
||||
): cv.string,
|
||||
vol.Required(
|
||||
CONF_API_KEY, default=user_input.get(CONF_API_KEY, "")
|
||||
): cv.string,
|
||||
vol.Required(
|
||||
CONF_FUEL_TYPES,
|
||||
default=user_input.get(CONF_FUEL_TYPES, list(FUEL_TYPES)),
|
||||
): cv.multi_select(FUEL_TYPES),
|
||||
vol.Required(
|
||||
CONF_LOCATION,
|
||||
default=user_input.get(
|
||||
CONF_LOCATION,
|
||||
{
|
||||
"latitude": self.hass.config.latitude,
|
||||
"longitude": self.hass.config.longitude,
|
||||
},
|
||||
),
|
||||
): selector({"location": {}}),
|
||||
vol.Required(
|
||||
CONF_RADIUS, default=user_input.get(CONF_RADIUS, DEFAULT_RADIUS)
|
||||
): selector(
|
||||
{
|
||||
"number": {
|
||||
"min": 0.1,
|
||||
"max": 25,
|
||||
"step": 0.1,
|
||||
CONF_UNIT_OF_MEASUREMENT: LENGTH_KILOMETERS,
|
||||
}
|
||||
}
|
||||
),
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
def _create_entry(
|
||||
self, data: dict[str, Any], options: dict[str, Any]
|
||||
) -> FlowResult:
|
||||
return self.async_create_entry(
|
||||
title=data[CONF_NAME],
|
||||
data=data,
|
||||
options=options,
|
||||
)
|
||||
|
||||
async def _get_nearby_stations(self, data: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Fetch nearby stations."""
|
||||
try:
|
||||
return await self.hass.async_add_executor_job(
|
||||
getNearbyStations,
|
||||
data[CONF_API_KEY],
|
||||
data[CONF_LOCATION][CONF_LATITUDE],
|
||||
data[CONF_LOCATION][CONF_LONGITUDE],
|
||||
data[CONF_RADIUS],
|
||||
"all",
|
||||
"dist",
|
||||
)
|
||||
except customException as err:
|
||||
return {"ok": False, "message": err, "exception": True}
|
||||
|
||||
|
||||
class OptionsFlowHandler(config_entries.OptionsFlow):
|
||||
"""Handle an options flow."""
|
||||
|
||||
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
|
||||
"""Initialize options flow."""
|
||||
self.config_entry = config_entry
|
||||
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle options flow."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="init",
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
CONF_SHOW_ON_MAP,
|
||||
default=self.config_entry.options[CONF_SHOW_ON_MAP],
|
||||
): bool,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
|
@ -6,4 +6,7 @@ NAME = "tankerkoenig"
|
|||
CONF_FUEL_TYPES = "fuel_types"
|
||||
CONF_STATIONS = "stations"
|
||||
|
||||
FUEL_TYPES = ["e5", "e10", "diesel"]
|
||||
DEFAULT_RADIUS = 2
|
||||
DEFAULT_SCAN_INTERVAL = 30
|
||||
|
||||
FUEL_TYPES = {"e5": "Super", "e10": "Super E10", "diesel": "Diesel"}
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
{
|
||||
"domain": "tankerkoenig",
|
||||
"name": "Tankerkoenig",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/tankerkoenig",
|
||||
"requirements": ["pytankerkoenig==0.0.6"],
|
||||
"codeowners": ["@guillempages"],
|
||||
"codeowners": ["@guillempages", "@mib1185"],
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["pytankerkoenig"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,22 +4,21 @@ from __future__ import annotations
|
|||
import logging
|
||||
|
||||
from homeassistant.components.sensor import STATE_CLASS_MEASUREMENT, SensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
ATTR_ATTRIBUTION,
|
||||
ATTR_ID,
|
||||
ATTR_LATITUDE,
|
||||
ATTR_LONGITUDE,
|
||||
CURRENCY_EURO,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
UpdateFailed,
|
||||
)
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN, NAME
|
||||
from . import TankerkoenigDataUpdateCoordinator
|
||||
from .const import DOMAIN, FUEL_TYPES
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -36,41 +35,20 @@ ATTRIBUTION = "Data provided by https://creativecommons.tankerkoenig.de"
|
|||
ICON = "mdi:gas-station"
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up the tankerkoenig sensors."""
|
||||
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
tankerkoenig = hass.data[DOMAIN]
|
||||
|
||||
async def async_update_data():
|
||||
"""Fetch data from API endpoint."""
|
||||
try:
|
||||
return await tankerkoenig.fetch_data()
|
||||
except LookupError as err:
|
||||
raise UpdateFailed("Failed to fetch data") from err
|
||||
|
||||
coordinator = DataUpdateCoordinator(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=NAME,
|
||||
update_method=async_update_data,
|
||||
update_interval=tankerkoenig.update_interval,
|
||||
)
|
||||
coordinator: TankerkoenigDataUpdateCoordinator = hass.data[DOMAIN][entry.unique_id]
|
||||
|
||||
# Fetch initial data so we have data when entities subscribe
|
||||
await coordinator.async_refresh()
|
||||
|
||||
stations = discovery_info.values()
|
||||
stations = coordinator.stations.values()
|
||||
entities = []
|
||||
for station in stations:
|
||||
for fuel in tankerkoenig.fuel_types:
|
||||
for fuel in coordinator.fuel_types:
|
||||
if fuel not in station:
|
||||
_LOGGER.warning(
|
||||
"Station %s does not offer %s fuel", station["id"], fuel
|
||||
|
|
@ -80,8 +58,7 @@ async def async_setup_platform(
|
|||
fuel,
|
||||
station,
|
||||
coordinator,
|
||||
f"{NAME}_{station['name']}_{fuel}",
|
||||
tankerkoenig.show_on_map,
|
||||
coordinator.show_on_map,
|
||||
)
|
||||
entities.append(sensor)
|
||||
_LOGGER.debug("Added sensors %s", entities)
|
||||
|
|
@ -94,26 +71,26 @@ class FuelPriceSensor(CoordinatorEntity, SensorEntity):
|
|||
|
||||
_attr_state_class = STATE_CLASS_MEASUREMENT
|
||||
|
||||
def __init__(self, fuel_type, station, coordinator, name, show_on_map):
|
||||
def __init__(self, fuel_type, station, coordinator, show_on_map):
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self._station = station
|
||||
self._station_id = station["id"]
|
||||
self._fuel_type = fuel_type
|
||||
self._name = name
|
||||
self._latitude = station["lat"]
|
||||
self._longitude = station["lng"]
|
||||
self._city = station["place"]
|
||||
self._house_number = station["houseNumber"]
|
||||
self._postcode = station["postCode"]
|
||||
self._street = station["street"]
|
||||
self._brand = self._station["brand"]
|
||||
self._price = station[fuel_type]
|
||||
self._show_on_map = show_on_map
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
return f"{self._brand} {self._street} {self._house_number} {FUEL_TYPES[self._fuel_type]}"
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
|
|
@ -136,6 +113,16 @@ class FuelPriceSensor(CoordinatorEntity, SensorEntity):
|
|||
"""Return a unique identifier for this entity."""
|
||||
return f"{self._station_id}_{self._fuel_type}"
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo | None:
|
||||
"""Return device info."""
|
||||
return DeviceInfo(
|
||||
identifiers={(ATTR_ID, self._station_id)},
|
||||
name=f"{self._brand} {self._street} {self._house_number}",
|
||||
model=self._brand,
|
||||
configuration_url="https://www.tankerkoenig.de",
|
||||
)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the attributes of the device."""
|
||||
|
|
|
|||
|
|
@ -0,0 +1,41 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"name": "Region name",
|
||||
"api_key": "[%key:common::config_flow::data::api_key%]",
|
||||
"fuel_types": "Fuel types",
|
||||
"location": "[%key:common::config_flow::data::location%]",
|
||||
"stations": "Additional fuel stations",
|
||||
"radius": "Search radius"
|
||||
}
|
||||
},
|
||||
"select_station":{
|
||||
"title": "Select stations to add",
|
||||
"description": "found {stations_count} stations in radius",
|
||||
"data": {
|
||||
"stations": "Stations"
|
||||
}
|
||||
}
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_location%]"
|
||||
},
|
||||
"error": {
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"no_stations": "Could not find any station in range."
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"title": "Tankerkoenig options",
|
||||
"data": {
|
||||
"scan_interval": "Update Interval",
|
||||
"show_on_map": "Show stations on map"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Location is already configured"
|
||||
},
|
||||
"error": {
|
||||
"invalid_auth": "Invalid authentication",
|
||||
"no_stations": "Could not find any station in range."
|
||||
},
|
||||
"step": {
|
||||
"select_station": {
|
||||
"data": {
|
||||
"stations": "Stations"
|
||||
},
|
||||
"description": "found {stations_count} stations in radius",
|
||||
"title": "Select stations to add"
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"api_key": "API Key",
|
||||
"fuel_types": "Fuel types",
|
||||
"location": "Location",
|
||||
"name": "Region name",
|
||||
"radius": "Search radius",
|
||||
"stations": "Additional fuel stations"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"data": {
|
||||
"scan_interval": "Update Interval",
|
||||
"show_on_map": "Show stations on map"
|
||||
},
|
||||
"title": "Tankerkoenig options"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -335,6 +335,7 @@ FLOWS = {
|
|||
"system_bridge",
|
||||
"tado",
|
||||
"tailscale",
|
||||
"tankerkoenig",
|
||||
"tasmota",
|
||||
"tellduslive",
|
||||
"tesla_wall_connector",
|
||||
|
|
|
|||
|
|
@ -1216,6 +1216,9 @@ pysqueezebox==0.5.5
|
|||
# homeassistant.components.syncthru
|
||||
pysyncthru==0.7.10
|
||||
|
||||
# homeassistant.components.tankerkoenig
|
||||
pytankerkoenig==0.0.6
|
||||
|
||||
# homeassistant.components.ecobee
|
||||
python-ecobee-api==0.2.14
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for Tankerkoenig component."""
|
||||
|
|
@ -0,0 +1,241 @@
|
|||
"""Tests for Tankerkoenig config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from pytankerkoenig import customException
|
||||
|
||||
from homeassistant.components.tankerkoenig.const import (
|
||||
CONF_FUEL_TYPES,
|
||||
CONF_STATIONS,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
|
||||
from homeassistant.const import (
|
||||
CONF_API_KEY,
|
||||
CONF_LATITUDE,
|
||||
CONF_LOCATION,
|
||||
CONF_LONGITUDE,
|
||||
CONF_NAME,
|
||||
CONF_RADIUS,
|
||||
CONF_SHOW_ON_MAP,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import (
|
||||
RESULT_TYPE_ABORT,
|
||||
RESULT_TYPE_CREATE_ENTRY,
|
||||
RESULT_TYPE_FORM,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_USER_DATA = {
|
||||
CONF_NAME: "Home",
|
||||
CONF_API_KEY: "269534f6-xxxx-xxxx-xxxx-yyyyzzzzxxxx",
|
||||
CONF_FUEL_TYPES: ["e5"],
|
||||
CONF_LOCATION: {CONF_LATITUDE: 51.0, CONF_LONGITUDE: 13.0},
|
||||
CONF_RADIUS: 2.0,
|
||||
}
|
||||
|
||||
MOCK_STATIONS_DATA = {
|
||||
CONF_STATIONS: [
|
||||
"3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8",
|
||||
"36b4b812-xxxx-xxxx-xxxx-c51735325858",
|
||||
],
|
||||
}
|
||||
|
||||
MOCK_IMPORT_DATA = {
|
||||
CONF_API_KEY: "269534f6-xxxx-xxxx-xxxx-yyyyzzzzxxxx",
|
||||
CONF_FUEL_TYPES: ["e5"],
|
||||
CONF_LOCATION: {CONF_LATITUDE: 51.0, CONF_LONGITUDE: 13.0},
|
||||
CONF_RADIUS: 2.0,
|
||||
CONF_STATIONS: [
|
||||
"3bcd61da-yyyy-yyyy-yyyy-19d5523a7ae8",
|
||||
"36b4b812-yyyy-yyyy-yyyy-c51735325858",
|
||||
],
|
||||
CONF_SHOW_ON_MAP: True,
|
||||
}
|
||||
|
||||
MOCK_NEARVY_STATIONS_OK = {
|
||||
"ok": True,
|
||||
"stations": [
|
||||
{
|
||||
"id": "3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8",
|
||||
"brand": "BrandA",
|
||||
"place": "CityA",
|
||||
"street": "Main",
|
||||
"houseNumber": "1",
|
||||
"dist": 1,
|
||||
},
|
||||
{
|
||||
"id": "36b4b812-xxxx-xxxx-xxxx-c51735325858",
|
||||
"brand": "BrandB",
|
||||
"place": "CityB",
|
||||
"street": "School",
|
||||
"houseNumber": "2",
|
||||
"dist": 2,
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
async def test_user(hass: HomeAssistant):
|
||||
"""Test starting a flow by user."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.tankerkoenig.async_setup_entry"
|
||||
) as mock_setup_entry, patch(
|
||||
"homeassistant.components.tankerkoenig.config_flow.getNearbyStations",
|
||||
return_value=MOCK_NEARVY_STATIONS_OK,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "select_station"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_STATIONS_DATA
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"][CONF_NAME] == "Home"
|
||||
assert result["data"][CONF_API_KEY] == "269534f6-xxxx-xxxx-xxxx-yyyyzzzzxxxx"
|
||||
assert result["data"][CONF_FUEL_TYPES] == ["e5"]
|
||||
assert result["data"][CONF_LOCATION] == {"latitude": 51.0, "longitude": 13.0}
|
||||
assert result["data"][CONF_RADIUS] == 2.0
|
||||
assert result["data"][CONF_STATIONS] == [
|
||||
"3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8",
|
||||
"36b4b812-xxxx-xxxx-xxxx-c51735325858",
|
||||
]
|
||||
assert result["options"][CONF_SHOW_ON_MAP]
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_setup_entry.called
|
||||
|
||||
|
||||
async def test_user_already_configured(hass: HomeAssistant):
|
||||
"""Test starting a flow by user with an already configured region."""
|
||||
|
||||
mock_config = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={**MOCK_USER_DATA, **MOCK_STATIONS_DATA},
|
||||
unique_id=f"{MOCK_USER_DATA[CONF_LOCATION][CONF_LATITUDE]}_{MOCK_USER_DATA[CONF_LOCATION][CONF_LONGITUDE]}",
|
||||
)
|
||||
mock_config.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
|
||||
|
||||
async def test_exception_security(hass: HomeAssistant):
|
||||
"""Test starting a flow by user with invalid api key."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.tankerkoenig.config_flow.getNearbyStations",
|
||||
side_effect=customException,
|
||||
):
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"][CONF_API_KEY] == "invalid_auth"
|
||||
|
||||
|
||||
async def test_user_no_stations(hass: HomeAssistant):
|
||||
"""Test starting a flow by user which does not find any station."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.tankerkoenig.config_flow.getNearbyStations",
|
||||
return_value={"ok": True, "stations": []},
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=MOCK_USER_DATA
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"][CONF_RADIUS] == "no_stations"
|
||||
|
||||
|
||||
async def test_import(hass: HomeAssistant):
|
||||
"""Test starting a flow by import."""
|
||||
with patch(
|
||||
"homeassistant.components.tankerkoenig.async_setup_entry"
|
||||
) as mock_setup_entry, patch(
|
||||
"homeassistant.components.tankerkoenig.config_flow.getNearbyStations",
|
||||
return_value=MOCK_NEARVY_STATIONS_OK,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=MOCK_IMPORT_DATA
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"][CONF_NAME] == "Home"
|
||||
assert result["data"][CONF_API_KEY] == "269534f6-xxxx-xxxx-xxxx-yyyyzzzzxxxx"
|
||||
assert result["data"][CONF_FUEL_TYPES] == ["e5"]
|
||||
assert result["data"][CONF_LOCATION] == {"latitude": 51.0, "longitude": 13.0}
|
||||
assert result["data"][CONF_RADIUS] == 2.0
|
||||
assert result["data"][CONF_STATIONS] == [
|
||||
"3bcd61da-xxxx-xxxx-xxxx-19d5523a7ae8",
|
||||
"36b4b812-xxxx-xxxx-xxxx-c51735325858",
|
||||
"3bcd61da-yyyy-yyyy-yyyy-19d5523a7ae8",
|
||||
"36b4b812-yyyy-yyyy-yyyy-c51735325858",
|
||||
]
|
||||
assert result["options"][CONF_SHOW_ON_MAP]
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_setup_entry.called
|
||||
|
||||
|
||||
async def test_options_flow(hass: HomeAssistant):
|
||||
"""Test options flow."""
|
||||
|
||||
mock_config = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=MOCK_USER_DATA,
|
||||
options={CONF_SHOW_ON_MAP: True},
|
||||
unique_id=f"{DOMAIN}_{MOCK_USER_DATA[CONF_LOCATION][CONF_LATITUDE]}_{MOCK_USER_DATA[CONF_LOCATION][CONF_LONGITUDE]}",
|
||||
)
|
||||
mock_config.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.tankerkoenig.async_setup_entry"
|
||||
) as mock_setup_entry:
|
||||
await mock_config.async_setup(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_setup_entry.called
|
||||
|
||||
result = await hass.config_entries.options.async_init(mock_config.entry_id)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "init"
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_SHOW_ON_MAP: False},
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert not mock_config.options[CONF_SHOW_ON_MAP]
|
||||
Loading…
Reference in New Issue