From 93fd85a48761287730182e3f75de3cd9db80c7fb Mon Sep 17 00:00:00 2001 From: SchiZzA Date: Mon, 11 May 2026 23:55:36 +0200 Subject: [PATCH] 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.py --- custom_components/sws12500/config_flow.py | 42 +--- custom_components/sws12500/const.py | 59 +++-- custom_components/sws12500/sensors_wslink.py | 39 ++- .../sws12500/translations/cs.json | 16 ++ .../sws12500/translations/en.json | 16 ++ custom_components/sws12500/utils.py | 58 +++-- custom_components/sws12500/windy_func.py | 13 +- tests/conftest.py | 38 +++ tests/test_t9_air_quality.py | 231 ++++++++++++++++++ 9 files changed, 419 insertions(+), 93 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_t9_air_quality.py diff --git a/custom_components/sws12500/config_flow.py b/custom_components/sws12500/config_flow.py index c3ceb0b..543f126 100644 --- a/custom_components/sws12500/config_flow.py +++ b/custom_components/sws12500/config_flow.py @@ -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", diff --git a/custom_components/sws12500/const.py b/custom_components/sws12500/const.py index e56727b..57f124e 100644 --- a/custom_components/sws12500/const.py +++ b/custom_components/sws12500/const.py @@ -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.""" diff --git a/custom_components/sws12500/sensors_wslink.py b/custom_components/sws12500/sensors_wslink.py index 7bfa6d5..7613827 100644 --- a/custom_components/sws12500/sensors_wslink.py +++ b/custom_components/sws12500/sensors_wslink.py @@ -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, + ), ) diff --git a/custom_components/sws12500/translations/cs.json b/custom_components/sws12500/translations/cs.json index bcd43bc..31d7d73 100644 --- a/custom_components/sws12500/translations/cs.json +++ b/custom_components/sws12500/translations/cs.json @@ -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": { diff --git a/custom_components/sws12500/translations/en.json b/custom_components/sws12500/translations/en.json index 9abdac8..3e9f553 100644 --- a/custom_components/sws12500/translations/en.json +++ b/custom_components/sws12500/translations/en.json @@ -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": { diff --git a/custom_components/sws12500/utils.py b/custom_components/sws12500/utils.py index b6f54ac..bbf913d 100644 --- a/custom_components/sws12500/utils.py +++ b/custom_components/sws12500/utils.py @@ -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) diff --git a/custom_components/sws12500/windy_func.py b/custom_components/sws12500/windy_func.py index 4d220c8..79be659 100644 --- a/custom_components/sws12500/windy_func.py +++ b/custom_components/sws12500/windy_func.py @@ -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 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..b73fb6b --- /dev/null +++ b/tests/conftest.py @@ -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//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) diff --git a/tests/test_t9_air_quality.py b/tests/test_t9_air_quality.py new file mode 100644 index 0000000..c6d3c2f --- /dev/null +++ b/tests/test_t9_air_quality.py @@ -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}