8000 Migrate nsw_fuel_station to config flow by buxtronix · Pull Request #146001 · home-assistant/core · GitHub
[go: up one dir, main page]
Skip to content

Migrate nsw_fuel_station to config flow #146001

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 10 commits into
base: dev
Choose a base branch
from
Draft

Migrate nsw_fuel_station to config flow #146001

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0991878
Migrate nsw_fuel_station to config flows
buxtronix May 22, 2025
5540a25
Migrate nsw_fuel to DataCoordinator
buxtronix May 26, 2025
5efd47b
Update unit tests
buxtronix Jun 1, 2025
25ee3e2
use multi_select
buxtronix Jun 2, 2025
4568077
Merge branch 'dev' into nsw_fuel_coord
buxtronix Jun 2, 2025
3895292
Address review comments.
buxtronix Jun 6, 2025
5000a9b
address comments
buxtronix Jun 6, 2025
7c667e6
Use consts in tests
buxtronix Jun 6, 2025
e084107
All fuel types, not just selected, plus other updates
buxtronix Jun 14, 2025
af9d5a1
Remove coordinator from config_flow
buxtronix Jun 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 64 additions & 40 deletions homeassistant/components/nsw_fuel_station/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,69 +4,93 @@

from dataclasses import dataclass
import datetime
import logging
from typing import TYPE_CHECKING

from nsw_fuel import FuelCheckClient, FuelCheckError, Station
from nsw_fuel import Station

from homeassistant.core import HomeAssistant
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import Platform
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

from .const import DATA_NSW_FUEL_STATION
from .const import DOMAIN
from .coordinator import NswFuelStationDataUpdateCoordinator

if TYPE_CHECKING:
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType

_LOGGER = logging.getLogger(__name__)

DOMAIN = "nsw_fuel_station"
SCAN_INTERVAL = datetime.timedelta(hours=1)

CONFIG_SCHEMA = cv.platform_only_config_schema(DOMAIN)

PLATFORMS: list[Platform] = [
Platform.SENSOR,
]


class FuelCheckData:
"""Holds a global coordinator."""

def __init__(self, hass: HomeAssistant) -> None:
"""Initialise the data."""
self.hass: HomeAssistant = hass
self.coordinator: NswFuelStationDataUpdateCoordinator = (
NswFuelStationDataUpdateCoordinator(
hass,
)
)


async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the NSW Fuel Station platform."""
client = FuelCheckClient()
fuel_data = hass.data.setdefault(DOMAIN, {})
if "coordinator" not in fuel_data:
fuel_data["coordinator"] = FuelCheckData(hass).coordinator
await fuel_data["coordinator"].async_config_entry_first_refresh()

if "sensor" not in config:
return True

for platform_config in config["sensor"]:
if platform_config["platform"] == DOMAIN:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=dict(platform_config), # Convert to dict
)
)

async def async_update_data():
return await hass.async_add_executor_job(fetch_station_price_data, client)
return True

coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
config_entry=None,
name="sensor",
update_interval=SCAN_INTERVAL,
update_method=async_update_data,
)
hass.data[DATA_NSW_FUEL_STATION] = coordinator

await coordinator.async_refresh()
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
) -> bool:
"""Set up this integration using UI."""
fuel_data = hass.data.setdefault(DOMAIN, {})
if "coordinator" not in fuel_data:
fuel_data["coordinator"] = FuelCheckData(hass).coordinator
await fuel_data["coordinator"].async_config_entry_first_refresh()
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

return True


async def async_unload_entry(
hass: HomeAssistant,
entry: ConfigEntry,
) -> bool:
"""Handle removal of an entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)


@dataclass
class StationPriceData:
"""Data structure for O(1) price and name lookups."""

stations: dict[int, Station]
prices: dict[tuple[int, str], float]


def fetch_station_price_data(client: FuelCheckClient) -> StationPriceData | None:
"""Fetch fuel price and station data."""
try:
raw_price_data = client.get_fuel_prices()
# Restructure prices and station details to be indexed by station code
# for O(1) lookup
return StationPriceData(
stations={s.code: s for s in raw_price_data.stations},
prices={
(p.station_code, p.fuel_type): p.price for p in raw_price_data.prices
},
)

except FuelCheckError as exc:
raise UpdateFailed(
f"Failed to fetch NSW Fuel station price data: {exc}"
) from exc
fuel_types: dict[str, str]
201 changes: 201 additions & 0 deletions homeassistant/components/nsw_fuel_station/config_flow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,201 @@
"""Adds config flow for NSW Fuel Check."""

from __future__ import annotations

from typing import Any

from nsw_fuel import Station
import voluptuous as vol

from homeassistant.config_entries import (
CONN_CLASS_CLOUD_POLL,
ConfigFlow,
ConfigFlowResult,
)
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)

from . import StationPriceData
from .const import (
CONF_FUEL_TYPES,
CONF_STATION_ID,
DOMAIN,
INPUT_FUEL_TYPES,
INPUT_SEARCH_TERM,
INPUT_STATION_ID,
)
from .coordinator import NswFuelStationDataUpdateCoordinator


class FuelCheckConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Fuel Check integration."""

VERSION = 1
CONNECTION_CLASS = CONN_CLASS_CLOUD_POLL

config_type: str | None
fuel_types: list[str]
stations: list[Station]
selected_station: Station
data: StationPriceData

async def _setup_coordinator(self):
coordinator = self.hass.data.get(DOMAIN, {}).get("coordinator")
if coordinator is None:
coordinator = NswFuelStationDataUpdateCoordinator(self.hass)
await coordinator.async_config_entry_first_refresh()
self.data = coordinator.data

async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Display the initial UI form."""
errors: dict[str, str] = {}

await self._setup_coordinator()
if self.data is None:
return self.async_abort(reason="fetch_failed")

if user_input is not None:
search_term = user_input[INPUT_SEARCH_TERM]
self.stations = [
station
for station in self.data.stations.values()
if (
search_term.lower() in station.name.lower()
or search_term.lower() in station.address.lower()
)
]
if not self.stations:
errors["base"] = "no_matching_stations"
else:
return await self.async_step_select_station()

return self.async_show_form(
step_id="user",
errors=errors,
data_schema=vol.Schema(
{
vol.Required(INPUT_SEARCH_TERM): str,
},
),
)

async def async_step_select_station(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show the station selector form."""
errors: dict[str, str] = {}

if user_input is not None:
station_id = int(user_input[INPUT_STATION_ID])
self.selected_station = next(
station for station in self.stations if station.code == station_id
)
for existing_entry in self._async_current_entries():
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you create an entry with a unique ID, you don't have to go through a loop. Try using code like this:

 await self.async_set_unique_id(YOUR_UNIQUE_STATION_ID)
 self._abort_if_unique_id_configured(updates=self._data)

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

if existing_entry.data[CONF_STATION_ID] == station_id:
errors["base"] = "station_exists"
if "base" not in errors:
return await self.async_step_select_fuel()

return self.async_show_form(
step_id="select_station",
errors=errors,
data_schema=vol.Schema(
{
vol.Required(INPUT_STATION_ID): SelectSelector(
SelectSelectorConfig(
options=[
SelectOptionDict(
value=str(station.code),
label=f"{station.name} - {station.address}",
)
for station in self.stations
],
mode=SelectSelectorMode.DROPDOWN,
)
),
},
),
)

async def async_step_select_fuel(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show the fuel type selection form."""
errors: dict[str, str] = {}

if user_input is not None:
if (
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No need to check here. Make the fuel type required in your schema - it will save this check.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

INPUT_FUEL_TYPES not in user_input
or len(user_input[INPUT_FUEL_TYPES]) < 1
):
errors["base"] = "missing_fuel_types"
else:
return self.async_create_entry(
title=self.selected_station.name,
data={
CONF_STATION_ID: self.selected_station.code,
CONF_FUEL_TYPES: user_input[INPUT_FUEL_TYPES],
},
)

valid_fuel_types = []
for station_code, fuel_type in self.data.prices:
if station_code == self.selected_station.code:
valid_fuel_types.append(
SelectOptionDict(
label=self.data.fuel_types.get(fuel_type, fuel_type),
value=fuel_type,
)
)

if len(valid_fuel_types) < 1:
return self.async_abort(reason="no_fuel_types")

return self.async_show_form(
step_id="select_fuel",
errors=errors,
data_schema=vol.Schema(
{
vol.Optional(INPUT_FUEL_TYPES): SelectSelector(
SelectSelectorConfig(
options=valid_fuel_types,
mode=SelectSelectorMode.DROPDOWN,
multiple=True,
)
),
}
),
)

async def async_step_import(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Import entries from yaml config."""
if not user_input:
return self.async_abort(reason="no_config")
station_id = user_input[INPUT_STATION_ID]
data = {
CONF_STATION_ID: station_id,
CONF_FUEL_TYPES: user_input[INPUT_FUEL_TYPES],
}
self._async_abort_entries_match({CONF_STATION_ID: station_id})

await self._setup_coordinator()
if self.data is None:
return self.async_abort(reason="fetch_failed")

station = self.data.stations.get(station_id)
name = "Unknown"
if station is not None:
name = station.name

return self.async_create_entry(
title=name,
data=data,
)
14 changes: 13 additions & 1 deletion homeassistant/components/nsw_fuel_station/const.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,15 @@
"""Constants for the NSW Fuel Station integration."""

DATA_NSW_FUEL_STATION = "nsw_fuel_station"
DOMAIN = "nsw_fuel_station"

INPUT_FUEL_TYPES = "fuel_types"
INPUT_STATION_ID = "station_id"
INPUT_SEARCH_TERM = "search_string"

ATTR_FUEL_TYPE = "fuel_type"
ATTR_STATION_ADDRESS = "station_address"
ATTR_STATION_ID = "station_id"
ATTR_STATION_NAME = "station_name"

CONF_STATION_ID = "station_id"
CONF_FUEL_TYPES = "fuel_types"
Loading
0