home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Decrease nsw fuel request volume (#49552)

Browse Source
This commit is contained in:
Nick Whyte
2021-05-29 11:53:20 +10:00
committed by GitHub
parent 8e87d638e1
commit 02cbb2025e
7 changed files with 218 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
homeassistant/components/nsw_fuel_station/__init__.py
View File

@ -1 +1,64 @@
"""The nsw_fuel_station component."""
from __future__ import annotations
from dataclasses import dataclass
import datetime
import logging
from nsw_fuel import FuelCheckClient, FuelCheckError, Station
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DATA_NSW_FUEL_STATION
_LOGGER = logging.getLogger(__name__)
DOMAIN = "nsw_fuel_station"
SCAN_INTERVAL = datetime.timedelta(hours=1)
async def async_setup(hass, config):
"""Set up the NSW Fuel Station platform."""
client = FuelCheckClient()
async def async_update_data():
return await hass.async_add_executor_job(fetch_station_price_data, client)
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name="sensor",
update_interval=SCAN_INTERVAL,
update_method=async_update_data,
)
hass.data[DATA_NSW_FUEL_STATION] = coordinator
await coordinator.async_refresh()
return True
@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:
_LOGGER.error("Failed to fetch NSW Fuel station price data. %s", exc)
return None

3
homeassistant/components/nsw_fuel_station/const.py Normal file
View File

@ -0,0 +1,3 @@
"""Constants for the NSW Fuel Station integration."""
DATA_NSW_FUEL_STATION = "nsw_fuel_station"

2
homeassistant/components/nsw_fuel_station/manifest.json
View File

@ -2,7 +2,7 @@
"domain": "nsw_fuel_station",
"name": "NSW Fuel Station Price",
"documentation": "https://www.home-assistant.io/integrations/nsw_fuel_station",
"requirements": ["nsw-fuel-api-client==1.0.10"],
"requirements": ["nsw-fuel-api-client==1.1.0"],
"codeowners": ["@nickw444"],
"iot_class": "cloud_polling"
}

160
homeassistant/components/nsw_fuel_station/sensor.py
View File

@ -1,16 +1,21 @@
"""Sensor platform to display the current fuel prices at a NSW fuel station."""
from __future__ import annotations
import datetime
import logging
from nsw_fuel import FuelCheckClient, FuelCheckError
import voluptuous as vol
from homeassistant.components.nsw_fuel_station import (
DATA_NSW_FUEL_STATION,
StationPriceData,
)
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.const import ATTR_ATTRIBUTION, CURRENCY_CENT, VOLUME_LITERS
import homeassistant.helpers.config_validation as cv
from homeassistant.util import Throttle
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
_LOGGER = logging.getLogger(__name__)
@ -35,7 +40,6 @@ CONF_ALLOWED_FUEL_TYPES = [
CONF_DEFAULT_FUEL_TYPES = ["E10", "U91"]
ATTRIBUTION = "Data provided by NSW Government FuelCheck"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_STATION_ID): cv.positive_int,
@ -45,11 +49,6 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
}
)
MIN_TIME_BETWEEN_UPDATES = datetime.timedelta(hours=1)
NOTIFICATION_ID = "nsw_fuel_station_notification"
NOTIFICATION_TITLE = "NSW Fuel Station Sensor Setup"
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the NSW Fuel Station sensor."""
@ -57,122 +56,63 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
station_id = config[CONF_STATION_ID]
fuel_types = config[CONF_FUEL_TYPES]
client = FuelCheckClient()
station_data = StationPriceData(client, station_id)
station_data.update()
coordinator = hass.data[DATA_NSW_FUEL_STATION]
if station_data.error is not None:
message = ("Error: {}. Check the logs for additional information.").format(
station_data.error
)
hass.components.persistent_notification.create(
message, title=NOTIFICATION_TITLE, notification_id=NOTIFICATION_ID
)
if coordinator.data is None:
_LOGGER.error("Initial fuel station price data not available")
return
available_fuel_types = station_data.get_available_fuel_types()
entities = []
for fuel_type in fuel_types:
if coordinator.data.prices.get((station_id, fuel_type)) is None:
_LOGGER.error(
"Fuel station price data not available for station %d and fuel type %s",
station_id,
fuel_type,
)
continue
add_entities(
[
StationPriceSensor(station_data, fuel_type)
for fuel_type in fuel_types
if fuel_type in available_fuel_types
]
)
entities.append(StationPriceSensor(coordinator, station_id, fuel_type))
add_entities(entities)
class StationPriceData:
"""An object to store and fetch the latest data for a given station."""
def __init__(self, client, station_id: int) -> None:
"""Initialize the sensor."""
self.station_id = station_id
self._client = client
self._data = None
self._reference_data = None
self.error = None
self._station_name = None
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Update the internal data using the API client."""
if self._reference_data is None:
try:
self._reference_data = self._client.get_reference_data()
except FuelCheckError as exc:
self.error = str(exc)
_LOGGER.error(
"Failed to fetch NSW Fuel station reference data. %s", exc
)
return
try:
self._data = self._client.get_fuel_prices_for_station(self.station_id)
except FuelCheckError as exc:
self.error = str(exc)
_LOGGER.error("Failed to fetch NSW Fuel station price data. %s", exc)
def for_fuel_type(self, fuel_type: str):
"""Return the price of the given fuel type."""
if self._data is None:
return None
return next(
(price for price in self._data if price.fuel_type == fuel_type), None
)
def get_available_fuel_types(self):
"""Return the available fuel types for the station."""
return [price.fuel_type for price in self._data]
def get_station_name(self) -> str:
"""Return the name of the station."""
if self._station_name is None:
name = None
if self._reference_data is not None:
name = next(
(
station.name
for station in self._reference_data.stations
if station.code == self.station_id
),
None,
)
self._station_name = name or f"station {self.station_id}"
return self._station_name
class StationPriceSensor(SensorEntity):
class StationPriceSensor(CoordinatorEntity, SensorEntity):
"""Implementation of a sensor that reports the fuel price for a station."""
def __init__(self, station_data: StationPriceData, fuel_type: str) -> None:
def __init__(
self,
coordinator: DataUpdateCoordinator[StationPriceData],
station_id: int,
fuel_type: str,
) -> None:
"""Initialize the sensor."""
self._station_data = station_data
super().__init__(coordinator)
self._station_id = station_id
self._fuel_type = fuel_type
@property
def name(self) -> str:
"""Return the name of the sensor."""
return f"{self._station_data.get_station_name()} {self._fuel_type}"
station_name = self._get_station_name()
return f"{station_name} {self._fuel_type}"
@property
def state(self) -> float | None:
"""Return the state of the sensor."""
price_info = self._station_data.for_fuel_type(self._fuel_type)
if price_info:
return price_info.price
if self.coordinator.data is None:
return None
return None
prices = self.coordinator.data.prices
return prices.get((self._station_id, self._fuel_type))
@property
def extra_state_attributes(self) -> dict:
"""Return the state attributes of the device."""
return {
ATTR_STATION_ID: self._station_data.station_id,
ATTR_STATION_NAME: self._station_data.get_station_name(),
ATTR_STATION_ID: self._station_id,
ATTR_STATION_NAME: self._get_station_name(),
ATTR_ATTRIBUTION: ATTRIBUTION,
}
@ -181,6 +121,18 @@ class StationPriceSensor(SensorEntity):
"""Return the units of measurement."""
return f"{CURRENCY_CENT}/{VOLUME_LITERS}"
def update(self):
"""Update current conditions."""
self._station_data.update()
def _get_station_name(self):
default_name = f"station {self._station_id}"
if self.coordinator.data is None:
return default_name
station = self.coordinator.data.stations.get(self._station_id)
if station is None:
return default_name
return station.name
@property
def unique_id(self) -> str | None:
"""Return a unique ID."""
return f"{self._station_id}_{self._fuel_type}"

2
requirements_all.txt
View File

@ -1024,7 +1024,7 @@ notify-events==1.0.4
nsapi==3.0.4
# homeassistant.components.nsw_fuel_station
nsw-fuel-api-client==1.0.10
nsw-fuel-api-client==1.1.0
# homeassistant.components.nuheat
nuheat==0.3.0

2
requirements_test_all.txt
View File

@ -560,7 +560,7 @@ nexia==0.9.7
notify-events==1.0.4
# homeassistant.components.nsw_fuel_station
nsw-fuel-api-client==1.0.10
nsw-fuel-api-client==1.1.0
# homeassistant.components.nuheat
nuheat==0.3.0

133
tests/components/nsw_fuel_station/test_sensor.py
View File

@ -1,7 +1,10 @@
"""The tests for the NSW Fuel Station sensor platform."""
from unittest.mock import patch
from nsw_fuel import FuelCheckError
from homeassistant.components import sensor
from homeassistant.components.nsw_fuel_station import DOMAIN
from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component
@ -12,6 +15,8 @@ VALID_CONFIG = {
"fuel_types": ["E10", "P95"],
}
VALID_CONFIG_EXPECTED_ENTITY_IDS = ["my_fake_station_p95", "my_fake_station_e10"]
class MockPrice:
"""Mock Price implementation."""
@ -34,68 +39,116 @@ class MockStation:
self.code = code
class MockGetReferenceDataResponse:
"""Mock GetReferenceDataResponse implementation."""
class MockGetFuelPricesResponse:
"""Mock GetFuelPricesResponse implementation."""
def __init__(self, stations):
"""Initialize a mock GetReferenceDataResponse instance."""
def __init__(self, prices, stations):
"""Initialize a mock GetFuelPricesResponse instance."""
self.prices = prices
self.stations = stations
class FuelCheckClientMock:
"""Mock FuelCheckClient implementation."""
def get_fuel_prices_for_station(self, station):
"""Return a fake fuel prices response."""
return [
MockPrice(
price=150.0,
fuel_type="P95",
last_updated=None,
price_unit=None,
station_code=350,
),
MockPrice(
price=140.0,
fuel_type="E10",
last_updated=None,
price_unit=None,
station_code=350,
),
]
def get_reference_data(self):
"""Return a fake reference data response."""
return MockGetReferenceDataResponse(
stations=[MockStation(code=350, name="My Fake Station")]
)
MOCK_FUEL_PRICES_RESPONSE = MockGetFuelPricesResponse(
prices=[
MockPrice(
price=150.0,
fuel_type="P95",
last_updated=None,
price_unit=None,
station_code=350,
),
MockPrice(
price=140.0,
fuel_type="E10",
last_updated=None,
price_unit=None,
station_code=350,
),
],
stations=[MockStation(code=350, name="My Fake Station")],
)
@patch(
"homeassistant.components.nsw_fuel_station.sensor.FuelCheckClient",
new=FuelCheckClientMock,
"homeassistant.components.nsw_fuel_station.FuelCheckClient.get_fuel_prices",
return_value=MOCK_FUEL_PRICES_RESPONSE,
)
async def test_setup(hass):
async def test_setup(get_fuel_prices, hass):
"""Test the setup with custom settings."""
assert await async_setup_component(hass, DOMAIN, {})
with assert_setup_component(1, sensor.DOMAIN):
assert await async_setup_component(
hass, sensor.DOMAIN, {"sensor": VALID_CONFIG}
)
await hass.async_block_till_done()
fake_entities = ["my_fake_station_p95", "my_fake_station_e10"]
for entity_id in fake_entities:
for entity_id in VALID_CONFIG_EXPECTED_ENTITY_IDS:
state = hass.states.get(f"sensor.{entity_id}")
assert state is not None
def raise_fuel_check_error():
"""Raise fuel check error for testing error cases."""
raise FuelCheckError()
@patch(
"homeassistant.components.nsw_fuel_station.sensor.FuelCheckClient",
new=FuelCheckClientMock,
"homeassistant.components.nsw_fuel_station.FuelCheckClient.get_fuel_prices",
side_effect=raise_fuel_check_error,
)
async def test_sensor_values(hass):
async def test_setup_error(get_fuel_prices, hass):
"""Test the setup with client throwing error."""
assert await async_setup_component(hass, DOMAIN, {})
with assert_setup_component(1, sensor.DOMAIN):
assert await async_setup_component(
hass, sensor.DOMAIN, {"sensor": VALID_CONFIG}
)
await hass.async_block_till_done()
for entity_id in VALID_CONFIG_EXPECTED_ENTITY_IDS:
state = hass.states.get(f"sensor.{entity_id}")
assert state is None
@patch(
"homeassistant.components.nsw_fuel_station.FuelCheckClient.get_fuel_prices",
return_value=MOCK_FUEL_PRICES_RESPONSE,
)
async def test_setup_error_no_station(get_fuel_prices, hass):
"""Test the setup with specified station not existing."""
assert await async_setup_component(hass, DOMAIN, {})
with assert_setup_component(2, sensor.DOMAIN):
assert await async_setup_component(
hass,
sensor.DOMAIN,
{
"sensor": [
{
"platform": "nsw_fuel_station",
"station_id": 350,
"fuel_types": ["E10"],
},
{
"platform": "nsw_fuel_station",
"station_id": 351,
"fuel_types": ["P95"],
},
]
},
)
await hass.async_block_till_done()
assert hass.states.get("sensor.my_fake_station_e10") is not None
assert hass.states.get("sensor.my_fake_station_p95") is None
@patch(
"homeassistant.components.nsw_fuel_station.FuelCheckClient.get_fuel_prices",
return_value=MOCK_FUEL_PRICES_RESPONSE,
)
async def test_sensor_values(get_fuel_prices, hass):
"""Test retrieval of sensor values."""
assert await async_setup_component(hass, DOMAIN, {})
assert await async_setup_component(hass, sensor.DOMAIN, {"sensor": VALID_CONFIG})
await hass.async_block_till_done()