Add HCHO / VOC air-quality sensors (T9 module)
Support the WSLink t9 air-quality module: - t9hcho (formaldehyde, ppb), - t9voclv (VOC level 1-5 -> ENUM state) - t9bat (0-5 battery -> percentage). The t9hcho/t9voclv/t9bat values are dropped when t9cn reports the module as disconnected, so no empty entities are created. - const: HCHO/VOC/T9_BATTERY keys, VOCLevel enum + VOC_LEVEL_MAP, BATTERY_NON_BINARY, CONNECTION_GATED_SENSORS, REMAP_WSLINK_ITEMS entries - utils: voc_level_to_text(), battery_5step_to_pct(), connection gating - sensors_wslink: HCHO / VOC / T9_BATTERY entity descriptions - translations (en, cs): hcho, voc (+ states), t9_battery - tests: tests/conftest.py + tests/test_t9_air_quality.pystable
SchiZzA
parent
8547e7a7f5
commit
93fd85a487
|
|
@ -4,12 +4,7 @@ from typing import Any
|
|||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import (
|
||||
ConfigEntry,
|
||||
ConfigFlow,
|
||||
ConfigFlowResult,
|
||||
OptionsFlow,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
|
||||
|
|
@ -79,9 +74,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
|
|||
|
||||
self.sensors = {
|
||||
SENSORS_TO_LOAD: (
|
||||
entry_data.get(SENSORS_TO_LOAD)
|
||||
if isinstance(entry_data.get(SENSORS_TO_LOAD), list)
|
||||
else []
|
||||
entry_data.get(SENSORS_TO_LOAD) if isinstance(entry_data.get(SENSORS_TO_LOAD), list) else []
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -93,14 +86,9 @@ class ConfigOptionsFlowHandler(OptionsFlow):
|
|||
}
|
||||
|
||||
self.windy_data_schema = {
|
||||
vol.Optional(
|
||||
WINDY_STATION_ID, default=self.windy_data.get(WINDY_STATION_ID, "")
|
||||
): str,
|
||||
vol.Optional(
|
||||
WINDY_STATION_PW, default=self.windy_data.get(WINDY_STATION_PW, "")
|
||||
): str,
|
||||
vol.Optional(WINDY_ENABLED, default=self.windy_data[WINDY_ENABLED]): bool
|
||||
or False,
|
||||
vol.Optional(WINDY_STATION_ID, default=self.windy_data.get(WINDY_STATION_ID, "")): str,
|
||||
vol.Optional(WINDY_STATION_PW, default=self.windy_data.get(WINDY_STATION_PW, "")): str,
|
||||
vol.Optional(WINDY_ENABLED, default=self.windy_data[WINDY_ENABLED]): bool or False,
|
||||
vol.Optional(
|
||||
WINDY_LOGGER_ENABLED,
|
||||
default=self.windy_data[WINDY_LOGGER_ENABLED],
|
||||
|
|
@ -116,19 +104,13 @@ class ConfigOptionsFlowHandler(OptionsFlow):
|
|||
}
|
||||
|
||||
self.pocasi_cz_schema = {
|
||||
vol.Required(
|
||||
POCASI_CZ_API_ID, default=self.pocasi_cz.get(POCASI_CZ_API_ID)
|
||||
): str,
|
||||
vol.Required(
|
||||
POCASI_CZ_API_KEY, default=self.pocasi_cz.get(POCASI_CZ_API_KEY)
|
||||
): str,
|
||||
vol.Required(POCASI_CZ_API_ID, default=self.pocasi_cz.get(POCASI_CZ_API_ID)): str,
|
||||
vol.Required(POCASI_CZ_API_KEY, default=self.pocasi_cz.get(POCASI_CZ_API_KEY)): str,
|
||||
vol.Required(
|
||||
POCASI_CZ_SEND_INTERVAL,
|
||||
default=self.pocasi_cz.get(POCASI_CZ_SEND_INTERVAL),
|
||||
): int,
|
||||
vol.Optional(
|
||||
POCASI_CZ_ENABLED, default=self.pocasi_cz.get(POCASI_CZ_ENABLED)
|
||||
): bool,
|
||||
vol.Optional(POCASI_CZ_ENABLED, default=self.pocasi_cz.get(POCASI_CZ_ENABLED)): bool,
|
||||
vol.Optional(
|
||||
POCASI_CZ_LOGGER_ENABLED,
|
||||
default=self.pocasi_cz.get(POCASI_CZ_LOGGER_ENABLED),
|
||||
|
|
@ -137,9 +119,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
|
|||
|
||||
async def async_step_init(self, user_input=None):
|
||||
"""Manage the options - show menu first."""
|
||||
return self.async_show_menu(
|
||||
step_id="init", menu_options=["basic", "windy", "pocasi"]
|
||||
)
|
||||
return self.async_show_menu(step_id="init", menu_options=["basic", "windy", "pocasi"])
|
||||
|
||||
async def async_step_basic(self, user_input=None):
|
||||
"""Manage basic options - credentials."""
|
||||
|
|
@ -293,9 +273,7 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
elif user_input[API_KEY] == user_input[API_ID]:
|
||||
errors["base"] = "valid_credentials_match"
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=DOMAIN, data=user_input, options=user_input
|
||||
)
|
||||
return self.async_create_entry(title=DOMAIN, data=user_input, options=user_input)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
|
|
|
|||
|
|
@ -28,26 +28,22 @@ POCASI_CZ_API_ID = "POCASI_CZ_API_ID"
|
|||
POCASI_CZ_SEND_INTERVAL = "POCASI_SEND_INTERVAL"
|
||||
POCASI_CZ_ENABLED = "pocasi_enabled_chcekbox"
|
||||
POCASI_CZ_LOGGER_ENABLED = "pocasi_logger_checkbox"
|
||||
POCASI_INVALID_KEY: Final = (
|
||||
"Pocasi Meteo refused to accept data. Invalid ID/Key combination?"
|
||||
)
|
||||
POCASI_INVALID_KEY: Final = "Pocasi Meteo refused to accept data. Invalid ID/Key combination?"
|
||||
POCASI_CZ_SUCCESS: Final = "Successfully sent data to Pocasi Meteo"
|
||||
POCASI_CZ_UNEXPECTED: Final = (
|
||||
"Pocasti Meteo responded unexpectedly 3 times in row. Resendig is now disabled!"
|
||||
)
|
||||
POCASI_CZ_UNEXPECTED: Final = "Pocasti Meteo responded unexpectedly 3 times in row. Resendig is now disabled!"
|
||||
|
||||
WINDY_STATION_ID = "WINDY_STATION_ID"
|
||||
WINDY_STATION_PW = "WINDY_STATION_PWD"
|
||||
WINDY_ENABLED: Final = "windy_enabled_checkbox"
|
||||
WINDY_LOGGER_ENABLED: Final = "windy_logger_checkbox"
|
||||
WINDY_NOT_INSERTED: Final = "Data was succefuly sent to Windy, but not inserted by Windy API. Does anyone else sent data to Windy?"
|
||||
WINDY_INVALID_KEY: Final = "Windy API KEY is invalid. Send data to Windy is now disabled. Check your API KEY and try again."
|
||||
WINDY_SUCCESS: Final = (
|
||||
"Windy successfully sent data and data was successfully inserted by Windy API"
|
||||
WINDY_NOT_INSERTED: Final = (
|
||||
"Data was succefuly sent to Windy, but not inserted by Windy API. Does anyone else sent data to Windy?"
|
||||
)
|
||||
WINDY_UNEXPECTED: Final = (
|
||||
"Windy responded unexpectedly 3 times in a row. Send to Windy is now disabled!"
|
||||
WINDY_INVALID_KEY: Final = (
|
||||
"Windy API KEY is invalid. Send data to Windy is now disabled. Check your API KEY and try again."
|
||||
)
|
||||
WINDY_SUCCESS: Final = "Windy successfully sent data and data was successfully inserted by Windy API"
|
||||
WINDY_UNEXPECTED: Final = "Windy responded unexpectedly 3 times in a row. Send to Windy is now disabled!"
|
||||
|
||||
INVALID_CREDENTIALS: Final = [
|
||||
"API",
|
||||
|
|
@ -118,6 +114,10 @@ CH4_CONNECTION: Final = "ch4_connection"
|
|||
HEAT_INDEX: Final = "heat_index"
|
||||
CHILL_INDEX: Final = "chill_index"
|
||||
WBGT_TEMP: Final = "wbgt_temp"
|
||||
HCHO: Final = "hcho"
|
||||
VOC: Final = "voc"
|
||||
T9_BATTERY: Final = "t9_battery" # T9 sensors are HCHO and VOC
|
||||
T9_CONN: Final = "t9_conn"
|
||||
|
||||
|
||||
REMAP_ITEMS: dict[str, str] = {
|
||||
|
|
@ -173,15 +173,11 @@ REMAP_WSLINK_ITEMS: dict[str, str] = {
|
|||
"inbat": INDOOR_BATTERY,
|
||||
"t234c1bat": CH2_BATTERY,
|
||||
"t1wbgt": WBGT_TEMP,
|
||||
"t9hcho": HCHO,
|
||||
"t9voclv": VOC,
|
||||
"t9bat": T9_BATTERY, # T9 battery is 0-5, where 5 is full
|
||||
}
|
||||
|
||||
# TODO: Add more sensors
|
||||
#
|
||||
# 'inbat' indoor battery level (1 normal, 0 low)
|
||||
# 't1bat': outdoor battery level (1 normal, 0 low)
|
||||
# 't234c1bat': CH2 battery level (1 normal, 0 low) CH2 in integration is CH1 in WSLink
|
||||
|
||||
|
||||
DISABLED_BY_DEFAULT: Final = [
|
||||
CH2_TEMP,
|
||||
CH2_HUMIDITY,
|
||||
|
|
@ -200,6 +196,31 @@ BATTERY_LIST = [
|
|||
CH2_BATTERY,
|
||||
]
|
||||
|
||||
BATTERY_NON_BINARY: list[str] = [T9_BATTERY]
|
||||
|
||||
CONNECTION_GATED_SENSORS: Final[dict[str, list[str]]] = {
|
||||
"t9cn": [HCHO, VOC, T9_BATTERY],
|
||||
}
|
||||
|
||||
|
||||
class VOCLevel(StrEnum):
|
||||
"""WSLink VOC Level 1-5 (1-worst)."""
|
||||
|
||||
UNHEALTHY = "unhealthy"
|
||||
POOR = "poor"
|
||||
MODERATE = "moderate"
|
||||
GOOD = "good"
|
||||
EXCELLENT = "excellent"
|
||||
|
||||
|
||||
VOC_LEVEL_MAP: dict[int, VOCLevel] = {
|
||||
1: VOCLevel.UNHEALTHY,
|
||||
2: VOCLevel.POOR,
|
||||
3: VOCLevel.MODERATE,
|
||||
4: VOCLevel.GOOD,
|
||||
5: VOCLevel.EXCELLENT,
|
||||
}
|
||||
|
||||
|
||||
class UnitOfDir(StrEnum):
|
||||
"""Wind direrction azimut."""
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ from typing import cast
|
|||
|
||||
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
|
||||
from homeassistant.const import (
|
||||
CONCENTRATION_PARTS_PER_BILLION,
|
||||
DEGREE,
|
||||
PERCENTAGE,
|
||||
UV_INDEX,
|
||||
|
|
@ -25,6 +26,7 @@ from .const import (
|
|||
CHILL_INDEX,
|
||||
DAILY_RAIN,
|
||||
DEW_POINT,
|
||||
HCHO,
|
||||
HEAT_INDEX,
|
||||
HOURLY_RAIN,
|
||||
INDOOR_BATTERY,
|
||||
|
|
@ -36,7 +38,9 @@ from .const import (
|
|||
OUTSIDE_TEMP,
|
||||
RAIN,
|
||||
SOLAR_RADIATION,
|
||||
T9_BATTERY,
|
||||
UV,
|
||||
VOC,
|
||||
WBGT_TEMP,
|
||||
WEEKLY_RAIN,
|
||||
WIND_AZIMUT,
|
||||
|
|
@ -45,9 +49,10 @@ from .const import (
|
|||
WIND_SPEED,
|
||||
YEARLY_RAIN,
|
||||
UnitOfDir,
|
||||
VOCLevel,
|
||||
)
|
||||
from .sensors_common import WeatherSensorEntityDescription
|
||||
from .utils import wind_dir_to_text
|
||||
from .utils import battery_5step_to_pct, voc_level_to_text, wind_dir_to_text
|
||||
|
||||
SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
|
||||
WeatherSensorEntityDescription(
|
||||
|
|
@ -311,21 +316,21 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
|
|||
translation_key=OUTSIDE_BATTERY,
|
||||
icon="mdi:battery-unknown",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
value_fn=lambda data: (data),
|
||||
value_fn=lambda data: data,
|
||||
),
|
||||
WeatherSensorEntityDescription(
|
||||
key=CH2_BATTERY,
|
||||
translation_key=CH2_BATTERY,
|
||||
icon="mdi:battery-unknown",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
value_fn=lambda data: (data),
|
||||
value_fn=lambda data: data,
|
||||
),
|
||||
WeatherSensorEntityDescription(
|
||||
key=INDOOR_BATTERY,
|
||||
translation_key=INDOOR_BATTERY,
|
||||
icon="mdi:battery-unknown",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
value_fn=lambda data: (data),
|
||||
value_fn=lambda data: data,
|
||||
),
|
||||
WeatherSensorEntityDescription(
|
||||
key=WBGT_TEMP,
|
||||
|
|
@ -337,4 +342,30 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
|
|||
suggested_display_precision=2,
|
||||
value_fn=lambda data: cast("int", data),
|
||||
),
|
||||
WeatherSensorEntityDescription(
|
||||
key=HCHO,
|
||||
translation_key=HCHO,
|
||||
device_class=SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS_PARTS,
|
||||
native_unit_of_measurement=CONCENTRATION_PARTS_PER_BILLION,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
icon="mdi:molecule",
|
||||
value_fn=lambda data: cast("int", data),
|
||||
),
|
||||
WeatherSensorEntityDescription(
|
||||
key=VOC,
|
||||
translation_key=VOC,
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
options=list(VOCLevel),
|
||||
icon="mdi:air-filter",
|
||||
value_fn=lambda data: cast("str", voc_level_to_text(data)),
|
||||
),
|
||||
WeatherSensorEntityDescription(
|
||||
key=T9_BATTERY,
|
||||
translation_key=T9_BATTERY,
|
||||
device_class=SensorDeviceClass.BATTERY,
|
||||
native_unit_of_measurement=PERCENTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=0,
|
||||
value_fn=battery_5step_to_pct,
|
||||
),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -189,6 +189,22 @@
|
|||
"wbgt_temp": {
|
||||
"name": "WBGT index"
|
||||
},
|
||||
"hcho": {
|
||||
"name": "Formaldehyd (HCHO)"
|
||||
},
|
||||
"voc": {
|
||||
"name": "Úroveň VOC",
|
||||
"state": {
|
||||
"unhealthy": "Nezdravá",
|
||||
"poor": "Špatná",
|
||||
"moderate": "Průměrná",
|
||||
"good": "Dobrá",
|
||||
"excellent": "Velmi dobrá"
|
||||
}
|
||||
},
|
||||
"t9_battery": {
|
||||
"name": "Baterie senzoru HCHO/VOC"
|
||||
},
|
||||
"wind_azimut": {
|
||||
"name": "Azimut",
|
||||
"state": {
|
||||
|
|
|
|||
|
|
@ -184,6 +184,22 @@
|
|||
"wbgt_index": {
|
||||
"name": "WBGT index"
|
||||
},
|
||||
"hcho": {
|
||||
"name": "Formaldehyde (HCHO)"
|
||||
},
|
||||
"voc": {
|
||||
"name": "VOC level",
|
||||
"state": {
|
||||
"unhealthy": "Unhealthy",
|
||||
"poor": "Poor",
|
||||
"moderate": "Moderate",
|
||||
"good": "Good",
|
||||
"excellent": "Excellent"
|
||||
}
|
||||
},
|
||||
"t9_battery": {
|
||||
"name": "HCHO/VOC sensor battery"
|
||||
},
|
||||
"wind_azimut": {
|
||||
"name": "Bearing",
|
||||
"state": {
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ from homeassistant.helpers.translation import async_get_translations
|
|||
|
||||
from .const import (
|
||||
AZIMUT,
|
||||
CONNECTION_GATED_SENSORS,
|
||||
DATABASE_PATH,
|
||||
DEV_DBG,
|
||||
OUTSIDE_HUMIDITY,
|
||||
|
|
@ -22,9 +23,11 @@ from .const import (
|
|||
REMAP_ITEMS,
|
||||
REMAP_WSLINK_ITEMS,
|
||||
SENSORS_TO_LOAD,
|
||||
VOC_LEVEL_MAP,
|
||||
WIND_SPEED,
|
||||
UnitOfBat,
|
||||
UnitOfDir,
|
||||
VOCLevel,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
|
@ -44,9 +47,7 @@ async def translations(
|
|||
|
||||
language = hass.config.language
|
||||
|
||||
_translations = await async_get_translations(
|
||||
hass, language, category, [translation_domain]
|
||||
)
|
||||
_translations = await async_get_translations(hass, language, category, [translation_domain])
|
||||
if localize_key in _translations:
|
||||
return _translations[localize_key]
|
||||
return ""
|
||||
|
|
@ -66,15 +67,11 @@ async def translated_notification(
|
|||
|
||||
localize_key = f"component.{translation_domain}.{category}.{translation_key}.{key}"
|
||||
|
||||
localize_title = (
|
||||
f"component.{translation_domain}.{category}.{translation_key}.title"
|
||||
)
|
||||
localize_title = f"component.{translation_domain}.{category}.{translation_key}.title"
|
||||
|
||||
language = hass.config.language
|
||||
|
||||
_translations = await async_get_translations(
|
||||
hass, language, category, [translation_domain]
|
||||
)
|
||||
_translations = await async_get_translations(hass, language, category, [translation_domain])
|
||||
if localize_key in _translations:
|
||||
if not translation_placeholders:
|
||||
persistent_notification.async_create(
|
||||
|
|
@ -85,14 +82,10 @@ async def translated_notification(
|
|||
)
|
||||
else:
|
||||
message = _translations[localize_key].format(**translation_placeholders)
|
||||
persistent_notification.async_create(
|
||||
hass, message, _translations[localize_title], notification_id
|
||||
)
|
||||
persistent_notification.async_create(hass, message, _translations[localize_title], notification_id)
|
||||
|
||||
|
||||
async def update_options(
|
||||
hass: HomeAssistant, entry: ConfigEntry, update_key, update_value
|
||||
) -> bool:
|
||||
async def update_options(hass: HomeAssistant, entry: ConfigEntry, update_key, update_value) -> bool:
|
||||
"""Update config.options entry."""
|
||||
conf = {**entry.options}
|
||||
conf[update_key] = update_value
|
||||
|
|
@ -128,6 +121,11 @@ def remap_wslink_items(entities):
|
|||
if item in REMAP_WSLINK_ITEMS:
|
||||
items[REMAP_WSLINK_ITEMS[item]] = entities[item]
|
||||
|
||||
for conn_key, gated in CONNECTION_GATED_SENSORS.items():
|
||||
if str(entities.get(conn_key, "0")) != "1":
|
||||
for key in gated:
|
||||
items.pop(key, None)
|
||||
|
||||
return items
|
||||
|
||||
|
||||
|
|
@ -137,9 +135,7 @@ def loaded_sensors(config_entry: ConfigEntry) -> list | None:
|
|||
return config_entry.options.get(SENSORS_TO_LOAD) or []
|
||||
|
||||
|
||||
def check_disabled(
|
||||
hass: HomeAssistant, items, config_entry: ConfigEntry
|
||||
) -> list | None:
|
||||
def check_disabled(hass: HomeAssistant, items, config_entry: ConfigEntry) -> list | None:
|
||||
"""Check if we have data for unloaded sensors.
|
||||
|
||||
If so, then add sensor to load queue.
|
||||
|
|
@ -284,11 +280,7 @@ def chill_index(data: Any, convert: bool = False) -> float | None:
|
|||
|
||||
return (
|
||||
round(
|
||||
(
|
||||
(35.7 + (0.6215 * temp))
|
||||
- (35.75 * (wind**0.16))
|
||||
+ (0.4275 * (temp * (wind**0.16)))
|
||||
),
|
||||
((35.7 + (0.6215 * temp)) - (35.75 * (wind**0.16)) + (0.4275 * (temp * (wind**0.16)))),
|
||||
2,
|
||||
)
|
||||
if temp < 50 and wind > 3
|
||||
|
|
@ -296,6 +288,22 @@ def chill_index(data: Any, convert: bool = False) -> float | None:
|
|||
)
|
||||
|
||||
|
||||
def voc_level_to_text(value: str) -> VOCLevel | None:
|
||||
"""Map 1-5 VOC level to text state."""
|
||||
if value in (None, ""):
|
||||
return None
|
||||
return VOC_LEVEL_MAP.get(int(value))
|
||||
|
||||
|
||||
def battery_5step_to_pct(value: str) -> int | None:
|
||||
"""Convert 0-5 battery steps to percentage."""
|
||||
|
||||
if value in (None, ""):
|
||||
return None
|
||||
|
||||
return round(int(value) / 5 * 100)
|
||||
|
||||
|
||||
def long_term_units_in_statistics_meta():
|
||||
"""Get units in long term statitstics."""
|
||||
sensor_units = []
|
||||
|
|
@ -314,9 +322,7 @@ def long_term_units_in_statistics_meta():
|
|||
"""
|
||||
)
|
||||
rows = db.fetchall()
|
||||
sensor_units = {
|
||||
statistic_id: f"{statistic_id} ({unit})" for statistic_id, unit in rows
|
||||
}
|
||||
sensor_units = {statistic_id: f"{statistic_id} ({unit})" for statistic_id, unit in rows}
|
||||
|
||||
except sqlite3.Error as e:
|
||||
_LOGGER.error("Error during data migration: %s", e)
|
||||
|
|
|
|||
|
|
@ -98,8 +98,6 @@ class WindyPush:
|
|||
from station. But we need to do some clean up.
|
||||
"""
|
||||
|
||||
text_for_test = None
|
||||
|
||||
if self.log:
|
||||
_LOGGER.info(
|
||||
"Windy last update = %s, next update at: %s",
|
||||
|
|
@ -169,9 +167,7 @@ class WindyPush:
|
|||
_LOGGER.info("Dataset for windy: %s", purged_data)
|
||||
session = async_get_clientsession(self.hass, verify_ssl=False)
|
||||
try:
|
||||
async with session.get(
|
||||
request_url, params=purged_data, headers=headers
|
||||
) as resp:
|
||||
async with session.get(request_url, params=purged_data, headers=headers) as resp:
|
||||
status = await resp.text()
|
||||
try:
|
||||
self.verify_windy_response(status)
|
||||
|
|
@ -179,26 +175,21 @@ class WindyPush:
|
|||
# log despite of settings
|
||||
_LOGGER.error(WINDY_NOT_INSERTED)
|
||||
|
||||
text_for_test = WINDY_NOT_INSERTED
|
||||
|
||||
except WindyApiKeyError:
|
||||
# log despite of settings
|
||||
_LOGGER.critical(WINDY_INVALID_KEY)
|
||||
text_for_test = WINDY_INVALID_KEY
|
||||
|
||||
await update_options(self.hass, self.config, WINDY_ENABLED, False)
|
||||
|
||||
except WindySuccess:
|
||||
if self.log:
|
||||
_LOGGER.info(WINDY_SUCCESS)
|
||||
text_for_test = WINDY_SUCCESS
|
||||
|
||||
except ClientError as ex:
|
||||
_LOGGER.critical("Invalid response from Windy: %s", str(ex))
|
||||
self.invalid_response_count += 1
|
||||
if self.invalid_response_count > 3:
|
||||
_LOGGER.critical(WINDY_UNEXPECTED)
|
||||
text_for_test = WINDY_UNEXPECTED
|
||||
await update_options(self.hass, self.config, WINDY_ENABLED, False)
|
||||
|
||||
self.last_update = datetime.now()
|
||||
|
|
@ -207,6 +198,4 @@ class WindyPush:
|
|||
if self.log:
|
||||
_LOGGER.info("Next update: %s", str(self.next_update))
|
||||
|
||||
if RESPONSE_FOR_TEST and text_for_test:
|
||||
return text_for_test
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -0,0 +1,38 @@
|
|||
"""Pytest configuration for tests under `dev/tests`.
|
||||
|
||||
Goals:
|
||||
- Make `custom_components.*` importable.
|
||||
- Keep this file lightweight and avoid global HA test-harness side effects.
|
||||
|
||||
Repository layout:
|
||||
- Root custom components: `SWS-12500/custom_components/...` (symlinked to `dev/custom_components/...`)
|
||||
- Integration sources: `SWS-12500/dev/custom_components/...`
|
||||
|
||||
Note:
|
||||
Some tests use lightweight `hass` stubs (e.g. SimpleNamespace) that are not compatible with
|
||||
Home Assistant's full test fixtures. Do NOT enable HA-only fixtures globally here.
|
||||
Instead, request such fixtures (e.g. `enable_custom_integrations`) explicitly in the specific
|
||||
tests that need HA's integration loader / flow managers.
|
||||
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
|
||||
def pytest_configure() -> None:
|
||||
"""Adjust sys.path so imports and HA loader discovery work in tests."""
|
||||
repo_root = Path(__file__).resolve().parents[2] # .../SWS-12500
|
||||
dev_root = repo_root / "dev"
|
||||
|
||||
# Ensure the repo root is importable so HA can find `custom_components/<domain>/manifest.json`.
|
||||
repo_root_str = str(repo_root)
|
||||
if repo_root_str not in sys.path:
|
||||
sys.path.insert(0, repo_root_str)
|
||||
|
||||
# Also ensure `dev/` is importable for direct imports from dev tooling/tests.
|
||||
dev_root_str = str(dev_root)
|
||||
if dev_root_str not in sys.path:
|
||||
sys.path.insert(0, dev_root_str)
|
||||
|
|
@ -0,0 +1,231 @@
|
|||
"""Tests for the T9 air-quality (HCHO / VOC) sensor support.
|
||||
|
||||
Covers what was added for the WSLink ``t9hcho`` / ``t9voclv`` / ``t9bat`` /
|
||||
``t9cn`` parameters:
|
||||
|
||||
- the new constants (``REMAP_WSLINK_ITEMS``, ``CONNECTION_GATED_SENSORS``,
|
||||
``BATTERY_NON_BINARY``, ``VOCLevel`` / ``VOC_LEVEL_MAP``)
|
||||
- the ``utils.voc_level_to_text`` and ``utils.battery_5step_to_pct`` helpers
|
||||
- the connection gating in ``utils.remap_wslink_items``
|
||||
- the new ``SENSOR_TYPES_WSLINK`` entity descriptions
|
||||
- the ``hcho`` / ``voc`` / ``t9_battery`` entries in the translation files
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from custom_components.sws12500.const import (
|
||||
BATTERY_LIST,
|
||||
BATTERY_NON_BINARY,
|
||||
CONNECTION_GATED_SENSORS,
|
||||
HCHO,
|
||||
OUTSIDE_TEMP,
|
||||
REMAP_WSLINK_ITEMS,
|
||||
T9_BATTERY,
|
||||
VOC,
|
||||
VOC_LEVEL_MAP,
|
||||
VOCLevel,
|
||||
)
|
||||
from custom_components.sws12500.sensors_wslink import SENSOR_TYPES_WSLINK
|
||||
from custom_components.sws12500.utils import battery_5step_to_pct, remap_wslink_items, voc_level_to_text
|
||||
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
|
||||
from homeassistant.const import CONCENTRATION_PARTS_PER_BILLION, PERCENTAGE
|
||||
|
||||
# Realistic WSLink payload taken from an issue report: the station sends every
|
||||
# parameter, even for channels with no sensor connected (``*cn == "0"``).
|
||||
ISSUE_PAYLOAD = {
|
||||
"rbar": "1013.3",
|
||||
"intem": "25.0",
|
||||
"inhum": "44",
|
||||
"t1cn": "1",
|
||||
"t1tem": "11.3",
|
||||
"t1hum": "92",
|
||||
"t234c1cn": "1",
|
||||
"t234c2cn": "1",
|
||||
"t234c3cn": "0",
|
||||
"t8cn": "0",
|
||||
"t9cn": "1",
|
||||
"t9hcho": "57",
|
||||
"t9voclv": "5",
|
||||
"t9bat": "5",
|
||||
"t10cn": "0",
|
||||
"t11cn": "0",
|
||||
"apiver": "1.00",
|
||||
}
|
||||
|
||||
|
||||
# --- constants -------------------------------------------------------------
|
||||
|
||||
|
||||
def test_t9_keys_are_remapped() -> None:
|
||||
assert REMAP_WSLINK_ITEMS["t9hcho"] == HCHO
|
||||
assert REMAP_WSLINK_ITEMS["t9voclv"] == VOC
|
||||
assert REMAP_WSLINK_ITEMS["t9bat"] == T9_BATTERY
|
||||
# t9cn is intentionally NOT remapped - it is only used as a gating flag.
|
||||
assert "t9cn" not in REMAP_WSLINK_ITEMS
|
||||
|
||||
|
||||
def test_connection_gated_sensors_definition() -> None:
|
||||
assert CONNECTION_GATED_SENSORS == {"t9cn": [HCHO, VOC, T9_BATTERY]}
|
||||
|
||||
|
||||
def test_t9_battery_is_non_binary_only() -> None:
|
||||
assert BATTERY_NON_BINARY == [T9_BATTERY]
|
||||
# the 0-5 / percentage battery must not be treated as a binary low/normal one
|
||||
assert T9_BATTERY not in BATTERY_LIST
|
||||
|
||||
|
||||
def test_voc_level_map_is_complete_and_ordered() -> None:
|
||||
# 1 == highest VOC reading (worst air) ... 5 == lowest VOC reading (best air)
|
||||
assert set(VOC_LEVEL_MAP) == {1, 2, 3, 4, 5}
|
||||
assert set(VOC_LEVEL_MAP.values()) == set(VOCLevel)
|
||||
assert VOC_LEVEL_MAP[1] is VOCLevel.UNHEALTHY
|
||||
assert VOC_LEVEL_MAP[5] is VOCLevel.EXCELENT
|
||||
assert [member.value for member in VOCLevel] == [
|
||||
"unhealthy",
|
||||
"poor",
|
||||
"moderate",
|
||||
"good",
|
||||
"excellent",
|
||||
]
|
||||
|
||||
|
||||
# --- voc_level_to_text -----------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("empty", [None, ""])
|
||||
def test_voc_level_to_text_handles_empty(empty) -> None:
|
||||
assert voc_level_to_text(empty) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raw", "expected"),
|
||||
[
|
||||
("1", VOCLevel.UNHEALTHY),
|
||||
("2", VOCLevel.POOR),
|
||||
("3", VOCLevel.MODERATE),
|
||||
("4", VOCLevel.GOOD),
|
||||
("5", VOCLevel.EXCELENT),
|
||||
(3, VOCLevel.MODERATE),
|
||||
],
|
||||
)
|
||||
def test_voc_level_to_text_maps_known_levels(raw, expected) -> None:
|
||||
assert voc_level_to_text(raw) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", ["0", "6", 0, 6])
|
||||
def test_voc_level_to_text_out_of_range_is_none(raw) -> None:
|
||||
assert voc_level_to_text(raw) is None
|
||||
|
||||
|
||||
# --- battery_5step_to_pct --------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("empty", [None, ""])
|
||||
def test_battery_5step_to_pct_handles_empty(empty) -> None:
|
||||
assert battery_5step_to_pct(empty) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raw", "expected"),
|
||||
[("0", 0), ("1", 20), ("2", 40), ("3", 60), ("4", 80), ("5", 100), (5, 100)],
|
||||
)
|
||||
def test_battery_5step_to_pct_scales_to_percentage(raw, expected) -> None:
|
||||
assert battery_5step_to_pct(raw) == expected
|
||||
|
||||
|
||||
# --- remap_wslink_items connection gating ----------------------------------
|
||||
|
||||
|
||||
def test_remap_keeps_t9_group_when_connected() -> None:
|
||||
out = remap_wslink_items({"t9cn": "1", "t9hcho": "57", "t9voclv": "5", "t9bat": "5", "t1tem": "11.3"})
|
||||
assert out[HCHO] == "57"
|
||||
assert out[VOC] == "5"
|
||||
assert out[T9_BATTERY] == "5"
|
||||
assert out[OUTSIDE_TEMP] == "11.3"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("conn", [{"t9cn": "0"}, {}], ids=["disconnected", "absent"])
|
||||
def test_remap_drops_t9_group_when_disconnected_or_absent(conn) -> None:
|
||||
out = remap_wslink_items({**conn, "t9hcho": "57", "t9voclv": "5", "t9bat": "5", "t1tem": "11.3"})
|
||||
assert HCHO not in out
|
||||
assert VOC not in out
|
||||
assert T9_BATTERY not in out
|
||||
# unrelated sensors are untouched by the gating
|
||||
assert out[OUTSIDE_TEMP] == "11.3"
|
||||
|
||||
|
||||
def test_remap_issue_payload_exposes_t9_when_connected() -> None:
|
||||
out = remap_wslink_items(ISSUE_PAYLOAD)
|
||||
# t9cn == "1" -> the T9 sensors are exposed
|
||||
assert out[HCHO] == "57"
|
||||
assert out[VOC] == "5"
|
||||
assert out[T9_BATTERY] == "5"
|
||||
# connection flags never leak into the sensor data
|
||||
assert "t9cn" not in out
|
||||
assert "t9_conn" not in out
|
||||
|
||||
|
||||
# --- sensor entity descriptions -------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wslink_descriptions():
|
||||
return {description.key: description for description in SENSOR_TYPES_WSLINK}
|
||||
|
||||
|
||||
def test_hcho_entity_description(wslink_descriptions) -> None:
|
||||
description = wslink_descriptions[HCHO]
|
||||
assert description.translation_key == HCHO
|
||||
assert description.device_class is SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS_PARTS
|
||||
assert description.native_unit_of_measurement == CONCENTRATION_PARTS_PER_BILLION
|
||||
assert description.state_class is SensorStateClass.MEASUREMENT
|
||||
# value_fn is a pass-through (typing.cast is a no-op at runtime; HA coerces the str)
|
||||
assert description.value_fn("57") == "57"
|
||||
|
||||
|
||||
def test_voc_entity_description(wslink_descriptions) -> None:
|
||||
description = wslink_descriptions[VOC]
|
||||
assert description.translation_key == VOC
|
||||
assert description.device_class is SensorDeviceClass.ENUM
|
||||
assert description.options == list(VOCLevel)
|
||||
# ENUM sensors must not declare a state_class
|
||||
assert description.state_class is None
|
||||
assert description.value_fn("1") == VOCLevel.UNHEALTHY
|
||||
assert description.value_fn("5") == "excellent"
|
||||
assert description.value_fn(None) is None
|
||||
|
||||
|
||||
def test_t9_battery_entity_description(wslink_descriptions) -> None:
|
||||
description = wslink_descriptions[T9_BATTERY]
|
||||
assert description.translation_key == T9_BATTERY
|
||||
assert description.device_class is SensorDeviceClass.BATTERY
|
||||
assert description.native_unit_of_measurement == PERCENTAGE
|
||||
assert description.state_class is SensorStateClass.MEASUREMENT
|
||||
assert description.suggested_display_precision == 0
|
||||
# no explicit icon -> HA renders the battery icon from the device class + %
|
||||
assert description.icon is None
|
||||
assert description.value_fn("5") == 100
|
||||
assert description.value_fn("0") == 0
|
||||
assert description.value_fn(None) is None
|
||||
|
||||
|
||||
# --- translation files -----------------------------------------------------
|
||||
|
||||
_TRANSLATIONS_DIR = Path(__file__).resolve().parents[1] / "custom_components" / "sws12500" / "translations"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename", ["en.json", "cs.json"])
|
||||
def test_translation_files_have_t9_entries(filename) -> None:
|
||||
sensors = json.loads((_TRANSLATIONS_DIR / filename).read_text(encoding="utf-8"))["entity"]["sensor"]
|
||||
|
||||
assert sensors["hcho"]["name"]
|
||||
assert sensors["t9_battery"]["name"]
|
||||
|
||||
voc = sensors["voc"]
|
||||
assert voc["name"]
|
||||
assert set(voc["state"]) == {member.value for member in VOCLevel}
|
||||
Loading…
Reference in New Issue