Refactor GIOS sensor platform (#89389)

This commit is contained in:
Maciej Bieniek 2023-03-28 08:36:42 +02:00 committed by GitHub
parent dc37d92197
commit 33fef5592f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 300 additions and 210 deletions

View file

@ -16,9 +16,6 @@ URL = "http://powietrze.gios.gov.pl/pjp/current/station_details/info/{station_id
API_TIMEOUT: Final = 30 API_TIMEOUT: Final = 30
ATTR_INDEX: Final = "index"
ATTR_STATION: Final = "station"
ATTR_C6H6: Final = "c6h6" ATTR_C6H6: Final = "c6h6"
ATTR_CO: Final = "co" ATTR_CO: Final = "co"
ATTR_NO2: Final = "no2" ATTR_NO2: Final = "no2"

View file

@ -4,7 +4,8 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
import logging import logging
from typing import Any, cast
from gios.model import GiosSensors
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
DOMAIN as PLATFORM, DOMAIN as PLATFORM,
@ -14,11 +15,7 @@ from homeassistant.components.sensor import (
SensorStateClass, SensorStateClass,
) )
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, CONF_NAME
ATTR_NAME,
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
CONF_NAME,
)
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import DeviceEntryType from homeassistant.helpers.device_registry import DeviceEntryType
@ -32,13 +29,11 @@ from .const import (
ATTR_AQI, ATTR_AQI,
ATTR_C6H6, ATTR_C6H6,
ATTR_CO, ATTR_CO,
ATTR_INDEX,
ATTR_NO2, ATTR_NO2,
ATTR_O3, ATTR_O3,
ATTR_PM10, ATTR_PM10,
ATTR_PM25, ATTR_PM25,
ATTR_SO2, ATTR_SO2,
ATTR_STATION,
ATTRIBUTION, ATTRIBUTION,
DOMAIN, DOMAIN,
MANUFACTURER, MANUFACTURER,
@ -49,17 +44,24 @@ _LOGGER = logging.getLogger(__name__)
@dataclass @dataclass
class GiosSensorEntityDescription(SensorEntityDescription): class GiosSensorRequiredKeysMixin:
"""Class for GIOS entity required keys."""
value: Callable[[GiosSensors], StateType]
@dataclass
class GiosSensorEntityDescription(SensorEntityDescription, GiosSensorRequiredKeysMixin):
"""Class describing GIOS sensor entities.""" """Class describing GIOS sensor entities."""
value: Callable | None = round subkey: str | None = None
SENSOR_TYPES: tuple[GiosSensorEntityDescription, ...] = ( SENSOR_TYPES: tuple[GiosSensorEntityDescription, ...] = (
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_AQI, key=ATTR_AQI,
name="AQI", name="AQI",
value=None, value=lambda sensors: sensors.aqi.value if sensors.aqi else None,
icon="mdi:air-filter", icon="mdi:air-filter",
device_class=SensorDeviceClass.ENUM, device_class=SensorDeviceClass.ENUM,
options=["very_bad", "bad", "sufficient", "moderate", "good", "very_good"], options=["very_bad", "bad", "sufficient", "moderate", "good", "very_good"],
@ -68,6 +70,8 @@ SENSOR_TYPES: tuple[GiosSensorEntityDescription, ...] = (
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_C6H6, key=ATTR_C6H6,
name="C6H6", name="C6H6",
value=lambda sensors: sensors.c6h6.value if sensors.c6h6 else None,
suggested_display_precision=0,
icon="mdi:molecule", icon="mdi:molecule",
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
@ -75,44 +79,107 @@ SENSOR_TYPES: tuple[GiosSensorEntityDescription, ...] = (
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_CO, key=ATTR_CO,
name="CO", name="CO",
value=lambda sensors: sensors.co.value if sensors.co else None,
suggested_display_precision=0,
icon="mdi:molecule",
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_NO2, key=ATTR_NO2,
name="NO2", name="NO2",
value=lambda sensors: sensors.no2.value if sensors.no2 else None,
suggested_display_precision=0,
device_class=SensorDeviceClass.NITROGEN_DIOXIDE, device_class=SensorDeviceClass.NITROGEN_DIOXIDE,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
GiosSensorEntityDescription(
key=ATTR_NO2,
subkey="index",
name="NO2 index",
value=lambda sensors: sensors.no2.index if sensors.no2 else None,
icon="mdi:molecule",
device_class=SensorDeviceClass.ENUM,
options=["very_bad", "bad", "sufficient", "moderate", "good", "very_good"],
translation_key="no2_index",
),
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_O3, key=ATTR_O3,
name="O3", name="O3",
value=lambda sensors: sensors.o3.value if sensors.o3 else None,
suggested_display_precision=0,
device_class=SensorDeviceClass.OZONE, device_class=SensorDeviceClass.OZONE,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
GiosSensorEntityDescription(
key=ATTR_O3,
subkey="index",
name="O3 index",
value=lambda sensors: sensors.o3.index if sensors.o3 else None,
icon="mdi:molecule",
device_class=SensorDeviceClass.ENUM,
options=["very_bad", "bad", "sufficient", "moderate", "good", "very_good"],
translation_key="o3_index",
),
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_PM10, key=ATTR_PM10,
name="PM10", name="PM10",
value=lambda sensors: sensors.pm10.value if sensors.pm10 else None,
suggested_display_precision=0,
device_class=SensorDeviceClass.PM10, device_class=SensorDeviceClass.PM10,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
GiosSensorEntityDescription(
key=ATTR_PM10,
subkey="index",
name="PM10 index",
value=lambda sensors: sensors.pm10.index if sensors.pm10 else None,
icon="mdi:molecule",
device_class=SensorDeviceClass.ENUM,
options=["very_bad", "bad", "sufficient", "moderate", "good", "very_good"],
translation_key="pm10_index",
),
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_PM25, key=ATTR_PM25,
name="PM2.5", name="PM2.5",
value=lambda sensors: sensors.pm25.value if sensors.pm25 else None,
suggested_display_precision=0,
device_class=SensorDeviceClass.PM25, device_class=SensorDeviceClass.PM25,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
GiosSensorEntityDescription(
key=ATTR_PM25,
subkey="index",
name="PM2.5 index",
value=lambda sensors: sensors.pm25.index if sensors.pm25 else None,
icon="mdi:molecule",
device_class=SensorDeviceClass.ENUM,
options=["very_bad", "bad", "sufficient", "moderate", "good", "very_good"],
translation_key="pm25_index",
),
GiosSensorEntityDescription( GiosSensorEntityDescription(
key=ATTR_SO2, key=ATTR_SO2,
name="SO2", name="SO2",
value=lambda sensors: sensors.so2.value if sensors.so2 else None,
suggested_display_precision=0,
device_class=SensorDeviceClass.SULPHUR_DIOXIDE, device_class=SensorDeviceClass.SULPHUR_DIOXIDE,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
), ),
GiosSensorEntityDescription(
key=ATTR_SO2,
subkey="index",
name="SO2 index",
value=lambda sensors: sensors.so2.index if sensors.so2 else None,
icon="mdi:molecule",
device_class=SensorDeviceClass.ENUM,
options=["very_bad", "bad", "sufficient", "moderate", "good", "very_good"],
translation_key="so2_index",
),
) )
@ -140,15 +207,13 @@ async def async_setup_entry(
) )
entity_registry.async_update_entity(entity_id, new_unique_id=new_unique_id) entity_registry.async_update_entity(entity_id, new_unique_id=new_unique_id)
sensors: list[GiosSensor | GiosAqiSensor] = [] sensors: list[GiosSensor] = []
for description in SENSOR_TYPES: for description in SENSOR_TYPES:
if getattr(coordinator.data, description.key) is None: if getattr(coordinator.data, description.key) is None:
continue continue
if description.key == ATTR_AQI: sensors.append(GiosSensor(name, coordinator, description))
sensors.append(GiosAqiSensor(name, coordinator, description))
else:
sensors.append(GiosSensor(name, coordinator, description))
async_add_entities(sensors) async_add_entities(sensors)
@ -174,45 +239,27 @@ class GiosSensor(CoordinatorEntity[GiosDataUpdateCoordinator], SensorEntity):
name=name, name=name,
configuration_url=URL.format(station_id=coordinator.gios.station_id), configuration_url=URL.format(station_id=coordinator.gios.station_id),
) )
self._attr_unique_id = f"{coordinator.gios.station_id}-{description.key}" if description.subkey:
self._attrs: dict[str, Any] = { self._attr_unique_id = (
ATTR_STATION: self.coordinator.gios.station_name, f"{coordinator.gios.station_id}-{description.key}-{description.subkey}"
} )
else:
self._attr_unique_id = f"{coordinator.gios.station_id}-{description.key}"
self.entity_description = description self.entity_description = description
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
self._attrs[ATTR_NAME] = getattr(
self.coordinator.data, self.entity_description.key
).name
self._attrs[ATTR_INDEX] = getattr(
self.coordinator.data, self.entity_description.key
).index
return self._attrs
@property @property
def native_value(self) -> StateType: def native_value(self) -> StateType:
"""Return the state.""" """Return the state."""
state = getattr(self.coordinator.data, self.entity_description.key).value return self.entity_description.value(self.coordinator.data)
assert self.entity_description.value is not None
return cast(StateType, self.entity_description.value(state))
class GiosAqiSensor(GiosSensor):
"""Define an GIOS AQI sensor."""
@property
def native_value(self) -> StateType:
"""Return the state."""
return cast(
StateType, getattr(self.coordinator.data, self.entity_description.key).value
)
@property @property
def available(self) -> bool: def available(self) -> bool:
"""Return if entity is available.""" """Return if entity is available."""
available = super().available available = super().available
return available and bool( sensor_data = getattr(self.coordinator.data, self.entity_description.key)
getattr(self.coordinator.data, self.entity_description.key)
) # Sometimes the API returns sensor data without indexes
if self.entity_description.subkey:
return available and bool(sensor_data.index)
return available and bool(sensor_data)

View file

@ -34,6 +34,56 @@
"good": "Good", "good": "Good",
"very_good": "Very good" "very_good": "Very good"
} }
},
"no2_index": {
"state": {
"very_bad": "[%key:component::gios::entity::sensor::aqi::state::very_bad%]",
"bad": "[%key:component::gios::entity::sensor::aqi::state::bad%]",
"sufficient": "[%key:component::gios::entity::sensor::aqi::state::sufficient%]",
"moderate": "[%key:component::gios::entity::sensor::aqi::state::moderate%]",
"good": "[%key:component::gios::entity::sensor::aqi::state::good%]",
"very_good": "[%key:component::gios::entity::sensor::aqi::state::very_good%]"
}
},
"o3_index": {
"state": {
"very_bad": "[%key:component::gios::entity::sensor::aqi::state::very_bad%]",
"bad": "[%key:component::gios::entity::sensor::aqi::state::bad%]",
"sufficient": "[%key:component::gios::entity::sensor::aqi::state::sufficient%]",
"moderate": "[%key:component::gios::entity::sensor::aqi::state::moderate%]",
"good": "[%key:component::gios::entity::sensor::aqi::state::good%]",
"very_good": "[%key:component::gios::entity::sensor::aqi::state::very_good%]"
}
},
"pm10_index": {
"state": {
"very_bad": "[%key:component::gios::entity::sensor::aqi::state::very_bad%]",
"bad": "[%key:component::gios::entity::sensor::aqi::state::bad%]",
"sufficient": "[%key:component::gios::entity::sensor::aqi::state::sufficient%]",
"moderate": "[%key:component::gios::entity::sensor::aqi::state::moderate%]",
"good": "[%key:component::gios::entity::sensor::aqi::state::good%]",
"very_good": "[%key:component::gios::entity::sensor::aqi::state::very_good%]"
}
},
"pm25_index": {
"state": {
"very_bad": "[%key:component::gios::entity::sensor::aqi::state::very_bad%]",
"bad": "[%key:component::gios::entity::sensor::aqi::state::bad%]",
"sufficient": "[%key:component::gios::entity::sensor::aqi::state::sufficient%]",
"moderate": "[%key:component::gios::entity::sensor::aqi::state::moderate%]",
"good": "[%key:component::gios::entity::sensor::aqi::state::good%]",
"very_good": "[%key:component::gios::entity::sensor::aqi::state::very_good%]"
}
},
"so2_index": {
"state": {
"very_bad": "[%key:component::gios::entity::sensor::aqi::state::very_bad%]",
"bad": "[%key:component::gios::entity::sensor::aqi::state::bad%]",
"sufficient": "[%key:component::gios::entity::sensor::aqi::state::sufficient%]",
"moderate": "[%key:component::gios::entity::sensor::aqi::state::moderate%]",
"good": "[%key:component::gios::entity::sensor::aqi::state::good%]",
"very_good": "[%key:component::gios::entity::sensor::aqi::state::very_good%]"
}
} }
} }
} }

View file

@ -5,12 +5,7 @@ from unittest.mock import patch
from gios import ApiError from gios import ApiError
from homeassistant.components.gios.const import ( from homeassistant.components.gios.const import ATTRIBUTION, DOMAIN
ATTR_INDEX,
ATTR_STATION,
ATTRIBUTION,
DOMAIN,
)
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
ATTR_OPTIONS, ATTR_OPTIONS,
ATTR_STATE_CLASS, ATTR_STATE_CLASS,
@ -42,16 +37,14 @@ async def test_sensor(hass: HomeAssistant) -> None:
state = hass.states.get("sensor.home_c6h6") state = hass.states.get("sensor.home_c6h6")
assert state assert state
assert state.state == "0" assert state.state == "0.23789"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert ( assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
) )
assert state.attributes.get(ATTR_ICON) == "mdi:molecule" assert state.attributes.get(ATTR_ICON) == "mdi:molecule"
assert state.attributes.get(ATTR_INDEX) == "very_good"
entry = registry.async_get("sensor.home_c6h6") entry = registry.async_get("sensor.home_c6h6")
assert entry assert entry
@ -59,16 +52,14 @@ async def test_sensor(hass: HomeAssistant) -> None:
state = hass.states.get("sensor.home_co") state = hass.states.get("sensor.home_co")
assert state assert state
assert state.state == "252" assert state.state == "251.874"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_DEVICE_CLASS) is None assert state.attributes.get(ATTR_DEVICE_CLASS) is None
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert ( assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
) )
assert state.attributes.get(ATTR_INDEX) == "good"
entry = registry.async_get("sensor.home_co") entry = registry.async_get("sensor.home_co")
assert entry assert entry
@ -76,94 +67,173 @@ async def test_sensor(hass: HomeAssistant) -> None:
state = hass.states.get("sensor.home_no2") state = hass.states.get("sensor.home_no2")
assert state assert state
assert state.state == "7" assert state.state == "7.13411"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.NITROGEN_DIOXIDE assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.NITROGEN_DIOXIDE
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert ( assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
) )
assert state.attributes.get(ATTR_INDEX) == "good"
entry = registry.async_get("sensor.home_no2") entry = registry.async_get("sensor.home_no2")
assert entry assert entry
assert entry.unique_id == "123-no2" assert entry.unique_id == "123-no2"
state = hass.states.get("sensor.home_no2_index")
assert state
assert state.state == "good"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.attributes.get(ATTR_OPTIONS) == [
"very_bad",
"bad",
"sufficient",
"moderate",
"good",
"very_good",
]
entry = registry.async_get("sensor.home_no2_index")
assert entry
assert entry.unique_id == "123-no2-index"
state = hass.states.get("sensor.home_o3") state = hass.states.get("sensor.home_o3")
assert state assert state
assert state.state == "96" assert state.state == "95.7768"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.OZONE assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.OZONE
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert ( assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
) )
assert state.attributes.get(ATTR_INDEX) == "good"
entry = registry.async_get("sensor.home_o3") entry = registry.async_get("sensor.home_o3")
assert entry assert entry
assert entry.unique_id == "123-o3" assert entry.unique_id == "123-o3"
state = hass.states.get("sensor.home_o3_index")
assert state
assert state.state == "good"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.attributes.get(ATTR_OPTIONS) == [
"very_bad",
"bad",
"sufficient",
"moderate",
"good",
"very_good",
]
entry = registry.async_get("sensor.home_o3_index")
assert entry
assert entry.unique_id == "123-o3-index"
state = hass.states.get("sensor.home_pm10") state = hass.states.get("sensor.home_pm10")
assert state assert state
assert state.state == "17" assert state.state == "16.8344"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.PM10 assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.PM10
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert ( assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
) )
assert state.attributes.get(ATTR_INDEX) == "good"
entry = registry.async_get("sensor.home_pm10") entry = registry.async_get("sensor.home_pm10")
assert entry assert entry
assert entry.unique_id == "123-pm10" assert entry.unique_id == "123-pm10"
state = hass.states.get("sensor.home_pm10_index")
assert state
assert state.state == "good"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.attributes.get(ATTR_OPTIONS) == [
"very_bad",
"bad",
"sufficient",
"moderate",
"good",
"very_good",
]
entry = registry.async_get("sensor.home_pm10_index")
assert entry
assert entry.unique_id == "123-pm10-index"
state = hass.states.get("sensor.home_pm2_5") state = hass.states.get("sensor.home_pm2_5")
assert state assert state
assert state.state == "4" assert state.state == "4"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.PM25 assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.PM25
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert ( assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
) )
assert state.attributes.get(ATTR_INDEX) == "good"
entry = registry.async_get("sensor.home_pm2_5") entry = registry.async_get("sensor.home_pm2_5")
assert entry assert entry
assert entry.unique_id == "123-pm25" assert entry.unique_id == "123-pm25"
state = hass.states.get("sensor.home_pm2_5_index")
assert state
assert state.state == "good"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.attributes.get(ATTR_OPTIONS) == [
"very_bad",
"bad",
"sufficient",
"moderate",
"good",
"very_good",
]
entry = registry.async_get("sensor.home_pm2_5_index")
assert entry
assert entry.unique_id == "123-pm25-index"
state = hass.states.get("sensor.home_so2") state = hass.states.get("sensor.home_so2")
assert state assert state
assert state.state == "4" assert state.state == "4.35478"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.SULPHUR_DIOXIDE assert state.attributes.get(ATTR_DEVICE_CLASS) == SensorDeviceClass.SULPHUR_DIOXIDE
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert ( assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
) )
assert state.attributes.get(ATTR_INDEX) == "very_good"
entry = registry.async_get("sensor.home_so2") entry = registry.async_get("sensor.home_so2")
assert entry assert entry
assert entry.unique_id == "123-so2" assert entry.unique_id == "123-so2"
state = hass.states.get("sensor.home_so2_index")
assert state
assert state.state == "very_good"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.attributes.get(ATTR_OPTIONS) == [
"very_bad",
"bad",
"sufficient",
"moderate",
"good",
"very_good",
]
entry = registry.async_get("sensor.home_so2_index")
assert entry
assert entry.unique_id == "123-so2-index"
state = hass.states.get("sensor.home_aqi") state = hass.states.get("sensor.home_aqi")
assert state assert state
assert state.state == "good" assert state.state == "good"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) is None assert state.attributes.get(ATTR_STATE_CLASS) is None
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.attributes.get(ATTR_OPTIONS) == [ assert state.attributes.get(ATTR_OPTIONS) == [
@ -182,13 +252,23 @@ async def test_sensor(hass: HomeAssistant) -> None:
async def test_availability(hass: HomeAssistant) -> None: async def test_availability(hass: HomeAssistant) -> None:
"""Ensure that we mark the entities unavailable correctly when service causes an error.""" """Ensure that we mark the entities unavailable correctly when service causes an error."""
indexes = json.loads(load_fixture("gios/indexes.json"))
sensors = json.loads(load_fixture("gios/sensors.json"))
await init_integration(hass) await init_integration(hass)
state = hass.states.get("sensor.home_pm2_5") state = hass.states.get("sensor.home_pm2_5")
assert state assert state
assert state.state != STATE_UNAVAILABLE
assert state.state == "4" assert state.state == "4"
state = hass.states.get("sensor.home_pm2_5_index")
assert state
assert state.state == "good"
state = hass.states.get("sensor.home_aqi")
assert state
assert state.state == "good"
future = utcnow() + timedelta(minutes=60) future = utcnow() + timedelta(minutes=60)
with patch( with patch(
"homeassistant.components.gios.Gios._get_all_sensors", "homeassistant.components.gios.Gios._get_all_sensors",
@ -201,10 +281,18 @@ async def test_availability(hass: HomeAssistant) -> None:
assert state assert state
assert state.state == STATE_UNAVAILABLE assert state.state == STATE_UNAVAILABLE
state = hass.states.get("sensor.home_pm2_5_index")
assert state
assert state.state == STATE_UNAVAILABLE
state = hass.states.get("sensor.home_aqi")
assert state
assert state.state == STATE_UNAVAILABLE
future = utcnow() + timedelta(minutes=120) future = utcnow() + timedelta(minutes=120)
with patch( with patch(
"homeassistant.components.gios.Gios._get_all_sensors", "homeassistant.components.gios.Gios._get_all_sensors",
return_value=json.loads(load_fixture("gios/sensors.json")), return_value=sensors,
), patch( ), patch(
"homeassistant.components.gios.Gios._get_indexes", "homeassistant.components.gios.Gios._get_indexes",
return_value={}, return_value={},
@ -214,161 +302,69 @@ async def test_availability(hass: HomeAssistant) -> None:
state = hass.states.get("sensor.home_pm2_5") state = hass.states.get("sensor.home_pm2_5")
assert state assert state
assert state.state != STATE_UNAVAILABLE
assert state.state == "4" assert state.state == "4"
# Indexes are empty so the state should be unavailable
state = hass.states.get("sensor.home_aqi") state = hass.states.get("sensor.home_aqi")
assert state assert state
assert state.state == STATE_UNAVAILABLE assert state.state == STATE_UNAVAILABLE
# Indexes are empty so the state should be unavailable
state = hass.states.get("sensor.home_pm2_5_index")
assert state
assert state.state == STATE_UNAVAILABLE
future = utcnow() + timedelta(minutes=180)
with patch(
"homeassistant.components.gios.Gios._get_all_sensors", return_value=sensors
), patch(
"homeassistant.components.gios.Gios._get_indexes",
return_value=indexes,
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get("sensor.home_pm2_5")
assert state
assert state.state == "4"
state = hass.states.get("sensor.home_pm2_5_index")
assert state
assert state.state == "good"
state = hass.states.get("sensor.home_aqi")
assert state
assert state.state == "good"
async def test_invalid_indexes(hass: HomeAssistant) -> None: async def test_invalid_indexes(hass: HomeAssistant) -> None:
"""Test states of the sensor when API returns invalid indexes.""" """Test states of the sensor when API returns invalid indexes."""
await init_integration(hass, invalid_indexes=True) await init_integration(hass, invalid_indexes=True)
registry = er.async_get(hass)
state = hass.states.get("sensor.home_c6h6") state = hass.states.get("sensor.home_no2_index")
assert state assert state
assert state.state == "0" assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes.get(ATTR_ICON) == "mdi:molecule"
assert state.attributes.get(ATTR_INDEX) is None
entry = registry.async_get("sensor.home_c6h6") state = hass.states.get("sensor.home_o3_index")
assert entry
assert entry.unique_id == "123-c6h6"
state = hass.states.get("sensor.home_co")
assert state assert state
assert state.state == "252" assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes.get(ATTR_INDEX) is None
entry = registry.async_get("sensor.home_co") state = hass.states.get("sensor.home_pm10_index")
assert entry
assert entry.unique_id == "123-co"
state = hass.states.get("sensor.home_no2")
assert state assert state
assert state.state == "7" assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes.get(ATTR_INDEX) is None
entry = registry.async_get("sensor.home_no2") state = hass.states.get("sensor.home_pm2_5_index")
assert entry
assert entry.unique_id == "123-no2"
state = hass.states.get("sensor.home_o3")
assert state assert state
assert state.state == "96" assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes.get(ATTR_INDEX) is None
entry = registry.async_get("sensor.home_o3") state = hass.states.get("sensor.home_so2_index")
assert entry
assert entry.unique_id == "123-o3"
state = hass.states.get("sensor.home_pm10")
assert state assert state
assert state.state == "17" assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes.get(ATTR_INDEX) is None
entry = registry.async_get("sensor.home_pm10")
assert entry
assert entry.unique_id == "123-pm10"
state = hass.states.get("sensor.home_pm2_5")
assert state
assert state.state == "4"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes.get(ATTR_INDEX) is None
entry = registry.async_get("sensor.home_pm2_5")
assert entry
assert entry.unique_id == "123-pm25"
state = hass.states.get("sensor.home_so2")
assert state
assert state.state == "4"
assert state.attributes.get(ATTR_ATTRIBUTION) == ATTRIBUTION
assert state.attributes.get(ATTR_STATION) == "Test Name 1"
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes.get(ATTR_INDEX) is None
entry = registry.async_get("sensor.home_so2")
assert entry
assert entry.unique_id == "123-so2"
state = hass.states.get("sensor.home_aqi") state = hass.states.get("sensor.home_aqi")
assert state is None assert state is None
async def test_aqi_sensor_availability(hass: HomeAssistant) -> None:
"""Ensure that we mark the AQI sensor unavailable correctly when indexes are invalid."""
await init_integration(hass)
state = hass.states.get("sensor.home_aqi")
assert state
assert state.state != STATE_UNAVAILABLE
assert state.state == "good"
future = utcnow() + timedelta(minutes=60)
with patch(
"homeassistant.components.gios.Gios._get_all_sensors",
return_value=json.loads(load_fixture("gios/sensors.json")),
), patch(
"homeassistant.components.gios.Gios._get_indexes",
return_value={},
):
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
state = hass.states.get("sensor.home_aqi")
assert state
assert state.state == STATE_UNAVAILABLE
async def test_unique_id_migration(hass: HomeAssistant) -> None: async def test_unique_id_migration(hass: HomeAssistant) -> None:
"""Test states of the unique_id migration.""" """Test states of the unique_id migration."""
registry = er.async_get(hass) registry = er.async_get(hass)