mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-29 21:03:40 +00:00
5a5c8ca3cc
- Skip asymmetry/zigzag rejection near the data tail and refactor spike validation so legitimate end-of-day spikes stop breaking periods. - Expose relaxation attempt sliders for both Best/Peak flows, wire the values through the coordinator, and extend the relaxation engine to honor the new max-attempt cap with richer logging & metadata. - Raise the default attempt count to eight flex levels so the 25% increment pattern can stretch much further before stopping, keeping translations and docs (including the matrix explanation) in sync across all locales. Impact: Tail spikes no longer get thrown out incorrectly, users can tune how aggressively the period search relaxes, and the defaults now find more viable periods on volatile days.
1013 lines
39 KiB
Python
1013 lines
39 KiB
Python
"""Adds config flow for tibber_prices."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, ClassVar
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.config_entries import (
|
|
ConfigEntry,
|
|
ConfigFlow,
|
|
ConfigFlowResult,
|
|
ConfigSubentryFlow,
|
|
OptionsFlow,
|
|
SubentryFlowResult,
|
|
)
|
|
from homeassistant.const import CONF_ACCESS_TOKEN
|
|
from homeassistant.core import callback
|
|
from homeassistant.helpers.aiohttp_client import async_create_clientsession
|
|
from homeassistant.helpers.selector import (
|
|
BooleanSelector,
|
|
NumberSelector,
|
|
NumberSelectorConfig,
|
|
NumberSelectorMode,
|
|
SelectOptionDict,
|
|
SelectSelector,
|
|
SelectSelectorConfig,
|
|
SelectSelectorMode,
|
|
TextSelector,
|
|
TextSelectorConfig,
|
|
TextSelectorType,
|
|
)
|
|
from homeassistant.loader import async_get_integration
|
|
|
|
from .api import (
|
|
TibberPricesApiClient,
|
|
TibberPricesApiClientAuthenticationError,
|
|
TibberPricesApiClientCommunicationError,
|
|
TibberPricesApiClientError,
|
|
)
|
|
from .const import (
|
|
BEST_PRICE_MAX_LEVEL_OPTIONS,
|
|
CONF_BEST_PRICE_FLEX,
|
|
CONF_BEST_PRICE_MAX_LEVEL,
|
|
CONF_BEST_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
CONF_BEST_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
CONF_BEST_PRICE_MIN_PERIOD_LENGTH,
|
|
CONF_ENABLE_MIN_PERIODS_BEST,
|
|
CONF_ENABLE_MIN_PERIODS_PEAK,
|
|
CONF_EXTENDED_DESCRIPTIONS,
|
|
CONF_MIN_PERIODS_BEST,
|
|
CONF_MIN_PERIODS_PEAK,
|
|
CONF_PEAK_PRICE_FLEX,
|
|
CONF_PEAK_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
CONF_PEAK_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
CONF_PEAK_PRICE_MIN_LEVEL,
|
|
CONF_PEAK_PRICE_MIN_PERIOD_LENGTH,
|
|
CONF_PRICE_RATING_THRESHOLD_HIGH,
|
|
CONF_PRICE_RATING_THRESHOLD_LOW,
|
|
CONF_PRICE_TREND_THRESHOLD_FALLING,
|
|
CONF_PRICE_TREND_THRESHOLD_RISING,
|
|
CONF_RELAXATION_ATTEMPTS_BEST,
|
|
CONF_RELAXATION_ATTEMPTS_PEAK,
|
|
CONF_RELAXATION_STEP_BEST,
|
|
CONF_RELAXATION_STEP_PEAK,
|
|
CONF_VOLATILITY_THRESHOLD_HIGH,
|
|
CONF_VOLATILITY_THRESHOLD_MODERATE,
|
|
CONF_VOLATILITY_THRESHOLD_VERY_HIGH,
|
|
DEFAULT_BEST_PRICE_FLEX,
|
|
DEFAULT_BEST_PRICE_MAX_LEVEL,
|
|
DEFAULT_BEST_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
DEFAULT_BEST_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
DEFAULT_BEST_PRICE_MIN_PERIOD_LENGTH,
|
|
DEFAULT_ENABLE_MIN_PERIODS_BEST,
|
|
DEFAULT_ENABLE_MIN_PERIODS_PEAK,
|
|
DEFAULT_EXTENDED_DESCRIPTIONS,
|
|
DEFAULT_MIN_PERIODS_BEST,
|
|
DEFAULT_MIN_PERIODS_PEAK,
|
|
DEFAULT_PEAK_PRICE_FLEX,
|
|
DEFAULT_PEAK_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
DEFAULT_PEAK_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
DEFAULT_PEAK_PRICE_MIN_LEVEL,
|
|
DEFAULT_PEAK_PRICE_MIN_PERIOD_LENGTH,
|
|
DEFAULT_PRICE_RATING_THRESHOLD_HIGH,
|
|
DEFAULT_PRICE_RATING_THRESHOLD_LOW,
|
|
DEFAULT_PRICE_TREND_THRESHOLD_FALLING,
|
|
DEFAULT_PRICE_TREND_THRESHOLD_RISING,
|
|
DEFAULT_RELAXATION_ATTEMPTS_BEST,
|
|
DEFAULT_RELAXATION_ATTEMPTS_PEAK,
|
|
DEFAULT_RELAXATION_STEP_BEST,
|
|
DEFAULT_RELAXATION_STEP_PEAK,
|
|
DEFAULT_VOLATILITY_THRESHOLD_HIGH,
|
|
DEFAULT_VOLATILITY_THRESHOLD_MODERATE,
|
|
DEFAULT_VOLATILITY_THRESHOLD_VERY_HIGH,
|
|
DOMAIN,
|
|
LOGGER,
|
|
PEAK_PRICE_MIN_LEVEL_OPTIONS,
|
|
)
|
|
|
|
|
|
class TibberPricesFlowHandler(ConfigFlow, domain=DOMAIN):
|
|
"""Config flow for tibber_prices."""
|
|
|
|
VERSION = 1
|
|
MINOR_VERSION = 0
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the config flow."""
|
|
super().__init__()
|
|
self._reauth_entry: ConfigEntry | None = None
|
|
self._viewer: dict | None = None
|
|
self._access_token: str | None = None
|
|
self._user_name: str | None = None
|
|
self._user_login: str | None = None
|
|
self._user_id: str | None = None
|
|
|
|
@classmethod
|
|
@callback
|
|
def async_get_supported_subentry_types(
|
|
cls,
|
|
config_entry: ConfigEntry, # noqa: ARG003
|
|
) -> dict[str, type[ConfigSubentryFlow]]:
|
|
"""Return subentries supported by this integration."""
|
|
return {"home": TibberPricesSubentryFlowHandler}
|
|
|
|
@staticmethod
|
|
@callback
|
|
def async_get_options_flow(_config_entry: ConfigEntry) -> OptionsFlow:
|
|
"""Create an options flow for this configentry."""
|
|
return TibberPricesOptionsFlowHandler()
|
|
|
|
def is_matching(self, other_flow: dict) -> bool:
|
|
"""Return True if match_dict matches this flow."""
|
|
return bool(other_flow.get("domain") == DOMAIN)
|
|
|
|
async def async_step_reauth(self, entry_data: dict[str, Any]) -> ConfigFlowResult: # noqa: ARG002
|
|
"""Handle reauth flow when access token becomes invalid."""
|
|
entry_id = self.context.get("entry_id")
|
|
if entry_id:
|
|
self._reauth_entry = self.hass.config_entries.async_get_entry(entry_id)
|
|
return await self.async_step_reauth_confirm()
|
|
|
|
async def async_step_reauth_confirm(self, user_input: dict | None = None) -> ConfigFlowResult:
|
|
"""Confirm reauth dialog - prompt for new access token."""
|
|
_errors = {}
|
|
|
|
if user_input is not None:
|
|
try:
|
|
viewer = await self._get_viewer_details(access_token=user_input[CONF_ACCESS_TOKEN])
|
|
except TibberPricesApiClientAuthenticationError as exception:
|
|
LOGGER.warning(exception)
|
|
_errors["base"] = "auth"
|
|
except TibberPricesApiClientCommunicationError as exception:
|
|
LOGGER.error(exception)
|
|
_errors["base"] = "connection"
|
|
except TibberPricesApiClientError as exception:
|
|
LOGGER.exception(exception)
|
|
_errors["base"] = "unknown"
|
|
else:
|
|
# Validate that the new token has access to all configured homes
|
|
if self._reauth_entry:
|
|
# Get all configured home IDs (main entry + subentries)
|
|
configured_home_ids = self._get_all_configured_home_ids(self._reauth_entry)
|
|
|
|
# Get accessible home IDs from the new token
|
|
accessible_homes = viewer.get("homes", [])
|
|
accessible_home_ids = {home["id"] for home in accessible_homes}
|
|
|
|
# Check if all configured homes are accessible with the new token
|
|
missing_home_ids = configured_home_ids - accessible_home_ids
|
|
|
|
if missing_home_ids:
|
|
# New token doesn't have access to all configured homes
|
|
LOGGER.error(
|
|
"New access token missing access to configured homes: %s",
|
|
", ".join(missing_home_ids),
|
|
)
|
|
_errors["base"] = "missing_homes"
|
|
else:
|
|
# Update the config entry with the new access token
|
|
self.hass.config_entries.async_update_entry(
|
|
self._reauth_entry,
|
|
data={
|
|
**self._reauth_entry.data,
|
|
CONF_ACCESS_TOKEN: user_input[CONF_ACCESS_TOKEN],
|
|
},
|
|
)
|
|
await self.hass.config_entries.async_reload(self._reauth_entry.entry_id)
|
|
return self.async_abort(reason="reauth_successful")
|
|
|
|
return self.async_show_form(
|
|
step_id="reauth_confirm",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Required(CONF_ACCESS_TOKEN): TextSelector(
|
|
TextSelectorConfig(type=TextSelectorType.TEXT),
|
|
),
|
|
}
|
|
),
|
|
errors=_errors,
|
|
)
|
|
|
|
async def async_step_user(
|
|
self,
|
|
user_input: dict | None = None,
|
|
) -> ConfigFlowResult:
|
|
"""Handle a flow initialized by the user. Only ask for access token."""
|
|
_errors = {}
|
|
if user_input is not None:
|
|
try:
|
|
viewer = await self._get_viewer_details(access_token=user_input[CONF_ACCESS_TOKEN])
|
|
except TibberPricesApiClientAuthenticationError as exception:
|
|
LOGGER.warning(exception)
|
|
_errors["base"] = "auth"
|
|
except TibberPricesApiClientCommunicationError as exception:
|
|
LOGGER.error(exception)
|
|
_errors["base"] = "connection"
|
|
except TibberPricesApiClientError as exception:
|
|
LOGGER.exception(exception)
|
|
_errors["base"] = "unknown"
|
|
else:
|
|
user_id = viewer.get("userId", None)
|
|
user_name = viewer.get("name") or user_id or "Unknown User"
|
|
user_login = viewer.get("login", "N/A")
|
|
homes = viewer.get("homes", [])
|
|
|
|
if not user_id:
|
|
LOGGER.error("No user ID found: %s", viewer)
|
|
return self.async_abort(reason="unknown")
|
|
|
|
if not homes:
|
|
LOGGER.error("No homes found: %s", viewer)
|
|
return self.async_abort(reason="unknown")
|
|
|
|
LOGGER.debug("Viewer data received: %s", viewer)
|
|
|
|
await self.async_set_unique_id(unique_id=str(user_id))
|
|
self._abort_if_unique_id_configured()
|
|
|
|
# Store viewer data in the flow for use in the next step
|
|
self._viewer = viewer
|
|
self._access_token = user_input[CONF_ACCESS_TOKEN]
|
|
self._user_name = user_name
|
|
self._user_login = user_login
|
|
self._user_id = user_id
|
|
|
|
# Move to home selection step
|
|
return await self.async_step_select_home()
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Required(
|
|
CONF_ACCESS_TOKEN,
|
|
default=(user_input or {}).get(CONF_ACCESS_TOKEN, vol.UNDEFINED),
|
|
): TextSelector(
|
|
TextSelectorConfig(
|
|
type=TextSelectorType.TEXT,
|
|
),
|
|
),
|
|
},
|
|
),
|
|
errors=_errors,
|
|
)
|
|
|
|
async def async_step_select_home(self, user_input: dict | None = None) -> ConfigFlowResult:
|
|
"""Handle home selection during initial setup."""
|
|
homes = self._viewer.get("homes", []) if self._viewer else []
|
|
|
|
if not homes:
|
|
return self.async_abort(reason="unknown")
|
|
|
|
if user_input is not None:
|
|
selected_home_id = user_input["home_id"]
|
|
selected_home = next((home for home in homes if home["id"] == selected_home_id), None)
|
|
|
|
if not selected_home:
|
|
return self.async_abort(reason="unknown")
|
|
|
|
data = {
|
|
CONF_ACCESS_TOKEN: self._access_token or "",
|
|
"home_id": selected_home_id,
|
|
"home_data": selected_home,
|
|
"homes": homes,
|
|
"user_login": self._user_login or "N/A",
|
|
}
|
|
|
|
return self.async_create_entry(
|
|
title=self._user_name or "Unknown User",
|
|
data=data,
|
|
description=f"{self._user_login} ({self._user_id})",
|
|
)
|
|
|
|
home_options = [
|
|
SelectOptionDict(
|
|
value=home["id"],
|
|
label=self._get_home_title(home),
|
|
)
|
|
for home in homes
|
|
]
|
|
|
|
return self.async_show_form(
|
|
step_id="select_home",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Required("home_id"): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=home_options,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
)
|
|
}
|
|
),
|
|
)
|
|
|
|
def _get_all_configured_home_ids(self, main_entry: ConfigEntry) -> set[str]:
|
|
"""Get all configured home IDs from main entry and all subentries."""
|
|
home_ids = set()
|
|
|
|
# Add home_id from main entry if it exists
|
|
if main_entry.data.get("home_id"):
|
|
home_ids.add(main_entry.data["home_id"])
|
|
|
|
# Add home_ids from all subentries
|
|
for entry in self.hass.config_entries.async_entries(DOMAIN):
|
|
if entry.data.get("home_id") and entry != main_entry:
|
|
home_ids.add(entry.data["home_id"])
|
|
|
|
return home_ids
|
|
|
|
@staticmethod
|
|
def _get_home_title(home: dict) -> str:
|
|
"""Generate a user-friendly title for a home."""
|
|
title = home.get("appNickname")
|
|
if title and title.strip():
|
|
return title.strip()
|
|
|
|
address = home.get("address", {})
|
|
if address:
|
|
parts = []
|
|
if address.get("address1"):
|
|
parts.append(address["address1"])
|
|
if address.get("city"):
|
|
parts.append(address["city"])
|
|
if parts:
|
|
return ", ".join(parts)
|
|
|
|
return home.get("id", "Unknown Home")
|
|
|
|
async def _get_viewer_details(self, access_token: str) -> dict:
|
|
"""Validate credentials and return information about the account (viewer object)."""
|
|
integration = await async_get_integration(self.hass, DOMAIN)
|
|
client = TibberPricesApiClient(
|
|
access_token=access_token,
|
|
session=async_create_clientsession(self.hass),
|
|
version=str(integration.version) if integration.version else "unknown",
|
|
)
|
|
result = await client.async_get_viewer_details()
|
|
return result["viewer"]
|
|
|
|
|
|
class TibberPricesSubentryFlowHandler(ConfigSubentryFlow):
|
|
"""Handle subentry flows for tibber_prices."""
|
|
|
|
async def async_step_user(self, user_input: dict[str, Any] | None = None) -> SubentryFlowResult:
|
|
"""User flow to add a new home."""
|
|
parent_entry = self._get_entry()
|
|
if not parent_entry or not hasattr(parent_entry, "runtime_data") or not parent_entry.runtime_data:
|
|
return self.async_abort(reason="no_parent_entry")
|
|
|
|
coordinator = parent_entry.runtime_data.coordinator
|
|
|
|
# Force refresh user data to get latest homes from Tibber API
|
|
await coordinator.refresh_user_data()
|
|
|
|
homes = coordinator.get_user_homes()
|
|
if not homes:
|
|
return self.async_abort(reason="no_available_homes")
|
|
|
|
if user_input is not None:
|
|
selected_home_id = user_input["home_id"]
|
|
selected_home = next((home for home in homes if home["id"] == selected_home_id), None)
|
|
|
|
if not selected_home:
|
|
return self.async_abort(reason="home_not_found")
|
|
|
|
home_title = self._get_home_title(selected_home)
|
|
home_id = selected_home["id"]
|
|
|
|
return self.async_create_entry(
|
|
title=home_title,
|
|
data={
|
|
"home_id": home_id,
|
|
"home_data": selected_home,
|
|
},
|
|
description=f"Subentry for {home_title}",
|
|
description_placeholders={"home_id": home_id},
|
|
unique_id=home_id,
|
|
)
|
|
|
|
# Get existing home IDs by checking all subentries for this parent
|
|
existing_home_ids = {
|
|
entry.data["home_id"]
|
|
for entry in self.hass.config_entries.async_entries(DOMAIN)
|
|
if entry.data.get("home_id") and entry != parent_entry
|
|
}
|
|
|
|
available_homes = [home for home in homes if home["id"] not in existing_home_ids]
|
|
|
|
if not available_homes:
|
|
return self.async_abort(reason="no_available_homes")
|
|
|
|
home_options = [
|
|
SelectOptionDict(
|
|
value=home["id"],
|
|
label=self._get_home_title(home),
|
|
)
|
|
for home in available_homes
|
|
]
|
|
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Required("home_id"): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=home_options,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
)
|
|
}
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=schema,
|
|
description_placeholders={},
|
|
errors={},
|
|
)
|
|
|
|
def _get_home_title(self, home: dict) -> str:
|
|
"""Generate a user-friendly title for a home."""
|
|
title = home.get("appNickname")
|
|
if title and title.strip():
|
|
return title.strip()
|
|
|
|
address = home.get("address", {})
|
|
if address:
|
|
parts = []
|
|
if address.get("address1"):
|
|
parts.append(address["address1"])
|
|
if address.get("city"):
|
|
parts.append(address["city"])
|
|
if parts:
|
|
return ", ".join(parts)
|
|
|
|
return home.get("id", "Unknown Home")
|
|
|
|
async def async_step_init(self, user_input: dict | None = None) -> SubentryFlowResult:
|
|
"""Manage the options for a subentry."""
|
|
subentry = self._get_reconfigure_subentry()
|
|
errors: dict[str, str] = {}
|
|
|
|
options = {
|
|
vol.Optional(
|
|
CONF_EXTENDED_DESCRIPTIONS,
|
|
default=subentry.data.get(CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS),
|
|
): BooleanSelector(),
|
|
}
|
|
|
|
if user_input is not None:
|
|
return self.async_update_and_abort(
|
|
self._get_entry(),
|
|
subentry,
|
|
data_updates=user_input,
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="init",
|
|
data_schema=vol.Schema(options),
|
|
errors=errors,
|
|
)
|
|
|
|
|
|
class TibberPricesOptionsFlowHandler(OptionsFlow):
|
|
"""Handle options for tibber_prices entries."""
|
|
|
|
# Step progress tracking
|
|
_TOTAL_STEPS: ClassVar[int] = 6
|
|
_STEP_INFO: ClassVar[dict[str, int]] = {
|
|
"init": 1,
|
|
"price_rating": 2,
|
|
"volatility": 3,
|
|
"best_price": 4,
|
|
"peak_price": 5,
|
|
"price_trend": 6,
|
|
}
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize options flow."""
|
|
self._options: dict[str, Any] = {}
|
|
|
|
def _get_step_description_placeholders(self, step_id: str) -> dict[str, str]:
|
|
"""Get description placeholders with step progress."""
|
|
if step_id not in self._STEP_INFO:
|
|
return {}
|
|
|
|
step_num = self._STEP_INFO[step_id]
|
|
|
|
# Get translations loaded by Home Assistant
|
|
standard_translations_key = f"{DOMAIN}_standard_translations_{self.hass.config.language}"
|
|
translations = self.hass.data.get(standard_translations_key, {})
|
|
|
|
# Get step progress text from translations with placeholders
|
|
step_progress_template = translations.get("common", {}).get("step_progress", "Step {step_num} of {total_steps}")
|
|
step_progress = step_progress_template.format(step_num=step_num, total_steps=self._TOTAL_STEPS)
|
|
|
|
return {
|
|
"step_progress": step_progress,
|
|
}
|
|
|
|
async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
|
"""Manage the options - General Settings."""
|
|
# Initialize options from config_entry on first call
|
|
if not self._options:
|
|
self._options = dict(self.config_entry.options)
|
|
|
|
if user_input is not None:
|
|
self._options.update(user_input)
|
|
return await self.async_step_price_rating()
|
|
|
|
return self.async_show_form(
|
|
step_id="init",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Optional(
|
|
CONF_EXTENDED_DESCRIPTIONS,
|
|
default=self.config_entry.options.get(
|
|
CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS
|
|
),
|
|
): BooleanSelector(),
|
|
}
|
|
),
|
|
description_placeholders={
|
|
**self._get_step_description_placeholders("init"),
|
|
"user_login": self.config_entry.data.get("user_login", "N/A"),
|
|
},
|
|
)
|
|
|
|
async def async_step_price_rating(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
|
"""Configure price rating thresholds."""
|
|
if user_input is not None:
|
|
self._options.update(user_input)
|
|
return await self.async_step_volatility()
|
|
|
|
return self.async_show_form(
|
|
step_id="price_rating",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Optional(
|
|
CONF_PRICE_RATING_THRESHOLD_LOW,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PRICE_RATING_THRESHOLD_LOW,
|
|
DEFAULT_PRICE_RATING_THRESHOLD_LOW,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=-100,
|
|
max=0,
|
|
unit_of_measurement="%",
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_PRICE_RATING_THRESHOLD_HIGH,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PRICE_RATING_THRESHOLD_HIGH,
|
|
DEFAULT_PRICE_RATING_THRESHOLD_HIGH,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0,
|
|
max=100,
|
|
unit_of_measurement="%",
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
}
|
|
),
|
|
description_placeholders=self._get_step_description_placeholders("price_rating"),
|
|
)
|
|
|
|
async def async_step_best_price(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
|
"""Configure best price period settings."""
|
|
if user_input is not None:
|
|
self._options.update(user_input)
|
|
return await self.async_step_peak_price()
|
|
|
|
return self.async_show_form(
|
|
step_id="best_price",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Optional(
|
|
CONF_BEST_PRICE_MIN_PERIOD_LENGTH,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_BEST_PRICE_MIN_PERIOD_LENGTH,
|
|
DEFAULT_BEST_PRICE_MIN_PERIOD_LENGTH,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=15,
|
|
max=240,
|
|
step=15,
|
|
unit_of_measurement="min",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_BEST_PRICE_FLEX,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_BEST_PRICE_FLEX,
|
|
DEFAULT_BEST_PRICE_FLEX,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0,
|
|
max=100,
|
|
step=1,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_BEST_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_BEST_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
DEFAULT_BEST_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0,
|
|
max=50,
|
|
step=1,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_BEST_PRICE_MAX_LEVEL,
|
|
default=self.config_entry.options.get(
|
|
CONF_BEST_PRICE_MAX_LEVEL,
|
|
DEFAULT_BEST_PRICE_MAX_LEVEL,
|
|
),
|
|
): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=BEST_PRICE_MAX_LEVEL_OPTIONS,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
translation_key="price_level",
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_BEST_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_BEST_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
DEFAULT_BEST_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0,
|
|
max=8,
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_ENABLE_MIN_PERIODS_BEST,
|
|
default=self.config_entry.options.get(
|
|
CONF_ENABLE_MIN_PERIODS_BEST,
|
|
DEFAULT_ENABLE_MIN_PERIODS_BEST,
|
|
),
|
|
): BooleanSelector(),
|
|
vol.Optional(
|
|
CONF_MIN_PERIODS_BEST,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_MIN_PERIODS_BEST,
|
|
DEFAULT_MIN_PERIODS_BEST,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=1,
|
|
max=10,
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_RELAXATION_STEP_BEST,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_RELAXATION_STEP_BEST,
|
|
DEFAULT_RELAXATION_STEP_BEST,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=5,
|
|
max=50,
|
|
step=5,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_RELAXATION_ATTEMPTS_BEST,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_RELAXATION_ATTEMPTS_BEST,
|
|
DEFAULT_RELAXATION_ATTEMPTS_BEST,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=1,
|
|
max=12,
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
}
|
|
),
|
|
description_placeholders=self._get_step_description_placeholders("best_price"),
|
|
)
|
|
|
|
async def async_step_peak_price(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
|
"""Configure peak price period settings."""
|
|
if user_input is not None:
|
|
self._options.update(user_input)
|
|
return await self.async_step_price_trend()
|
|
|
|
return self.async_show_form(
|
|
step_id="peak_price",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Optional(
|
|
CONF_PEAK_PRICE_MIN_PERIOD_LENGTH,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PEAK_PRICE_MIN_PERIOD_LENGTH,
|
|
DEFAULT_PEAK_PRICE_MIN_PERIOD_LENGTH,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=15,
|
|
max=240,
|
|
step=15,
|
|
unit_of_measurement="min",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_PEAK_PRICE_FLEX,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PEAK_PRICE_FLEX,
|
|
DEFAULT_PEAK_PRICE_FLEX,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=-100,
|
|
max=0,
|
|
step=1,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_PEAK_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PEAK_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
DEFAULT_PEAK_PRICE_MIN_DISTANCE_FROM_AVG,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0,
|
|
max=50,
|
|
step=1,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_PEAK_PRICE_MIN_LEVEL,
|
|
default=self.config_entry.options.get(
|
|
CONF_PEAK_PRICE_MIN_LEVEL,
|
|
DEFAULT_PEAK_PRICE_MIN_LEVEL,
|
|
),
|
|
): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=PEAK_PRICE_MIN_LEVEL_OPTIONS,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
translation_key="price_level",
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_PEAK_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PEAK_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
DEFAULT_PEAK_PRICE_MAX_LEVEL_GAP_COUNT,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0,
|
|
max=8,
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_ENABLE_MIN_PERIODS_PEAK,
|
|
default=self.config_entry.options.get(
|
|
CONF_ENABLE_MIN_PERIODS_PEAK,
|
|
DEFAULT_ENABLE_MIN_PERIODS_PEAK,
|
|
),
|
|
): BooleanSelector(),
|
|
vol.Optional(
|
|
CONF_MIN_PERIODS_PEAK,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_MIN_PERIODS_PEAK,
|
|
DEFAULT_MIN_PERIODS_PEAK,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=1,
|
|
max=10,
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_RELAXATION_STEP_PEAK,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_RELAXATION_STEP_PEAK,
|
|
DEFAULT_RELAXATION_STEP_PEAK,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=5,
|
|
max=50,
|
|
step=5,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_RELAXATION_ATTEMPTS_PEAK,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_RELAXATION_ATTEMPTS_PEAK,
|
|
DEFAULT_RELAXATION_ATTEMPTS_PEAK,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=1,
|
|
max=12,
|
|
step=1,
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
}
|
|
),
|
|
description_placeholders=self._get_step_description_placeholders("peak_price"),
|
|
)
|
|
|
|
async def async_step_price_trend(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
|
"""Configure price trend thresholds."""
|
|
if user_input is not None:
|
|
self._options.update(user_input)
|
|
return self.async_create_entry(title="", data=self._options)
|
|
|
|
return self.async_show_form(
|
|
step_id="price_trend",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Optional(
|
|
CONF_PRICE_TREND_THRESHOLD_RISING,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PRICE_TREND_THRESHOLD_RISING,
|
|
DEFAULT_PRICE_TREND_THRESHOLD_RISING,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=1,
|
|
max=50,
|
|
step=1,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_PRICE_TREND_THRESHOLD_FALLING,
|
|
default=int(
|
|
self.config_entry.options.get(
|
|
CONF_PRICE_TREND_THRESHOLD_FALLING,
|
|
DEFAULT_PRICE_TREND_THRESHOLD_FALLING,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=-50,
|
|
max=-1,
|
|
step=1,
|
|
unit_of_measurement="%",
|
|
mode=NumberSelectorMode.SLIDER,
|
|
),
|
|
),
|
|
}
|
|
),
|
|
description_placeholders=self._get_step_description_placeholders("price_trend"),
|
|
)
|
|
|
|
async def async_step_volatility(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
|
"""Configure volatility thresholds and period filtering."""
|
|
if user_input is not None:
|
|
self._options.update(user_input)
|
|
return await self.async_step_best_price()
|
|
|
|
return self.async_show_form(
|
|
step_id="volatility",
|
|
data_schema=vol.Schema(
|
|
{
|
|
vol.Optional(
|
|
CONF_VOLATILITY_THRESHOLD_MODERATE,
|
|
default=float(
|
|
self.config_entry.options.get(
|
|
CONF_VOLATILITY_THRESHOLD_MODERATE,
|
|
DEFAULT_VOLATILITY_THRESHOLD_MODERATE,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0.0,
|
|
max=100.0,
|
|
step=0.1,
|
|
unit_of_measurement="ct",
|
|
mode=NumberSelectorMode.BOX,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_VOLATILITY_THRESHOLD_HIGH,
|
|
default=float(
|
|
self.config_entry.options.get(
|
|
CONF_VOLATILITY_THRESHOLD_HIGH,
|
|
DEFAULT_VOLATILITY_THRESHOLD_HIGH,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0.0,
|
|
max=100.0,
|
|
step=0.1,
|
|
unit_of_measurement="ct",
|
|
mode=NumberSelectorMode.BOX,
|
|
),
|
|
),
|
|
vol.Optional(
|
|
CONF_VOLATILITY_THRESHOLD_VERY_HIGH,
|
|
default=float(
|
|
self.config_entry.options.get(
|
|
CONF_VOLATILITY_THRESHOLD_VERY_HIGH,
|
|
DEFAULT_VOLATILITY_THRESHOLD_VERY_HIGH,
|
|
)
|
|
),
|
|
): NumberSelector(
|
|
NumberSelectorConfig(
|
|
min=0.0,
|
|
max=100.0,
|
|
step=0.1,
|
|
unit_of_measurement="ct",
|
|
mode=NumberSelectorMode.BOX,
|
|
),
|
|
),
|
|
}
|
|
),
|
|
description_placeholders=self._get_step_description_placeholders("volatility"),
|
|
)
|