From 21dfa61cd5424aabf64157ddff8d7440b9f17288 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/const.py | 32 +++ custom_components/sws12500/sensors_wslink.py | 31 +++ .../sws12500/translations/cs.json | 16 ++ .../sws12500/translations/en.json | 16 ++ custom_components/sws12500/utils.py | 33 +-- tests/test_t9_air_quality.py | 231 ++++++++++++++++++ 6 files changed, 336 insertions(+), 23 deletions(-) create mode 100644 tests/test_t9_air_quality.py diff --git a/custom_components/sws12500/const.py b/custom_components/sws12500/const.py index f597c20..266826e 100644 --- a/custom_components/sws12500/const.py +++ b/custom_components/sws12500/const.py @@ -83,6 +83,10 @@ CH8_BATTERY: Final = "ch8_battery" 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" # Health specific constants @@ -380,6 +384,9 @@ REMAP_WSLINK_ITEMS: dict[str, str] = { "t234c6bat": CH7_BATTERY, "t234c7bat": CH8_BATTERY, "t1wbgt": WBGT_TEMP, + "t9hcho": HCHO, + "t9voclv": VOC, + "t9bat": T9_BATTERY, # T9 battery is 0-5, where 5 is full } # NOTE: Add more sensors @@ -481,6 +488,31 @@ BATTERY_LIST = [ CH8_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 8500e35..d1056d7 100644 --- a/custom_components/sws12500/sensors_wslink.py +++ b/custom_components/sws12500/sensors_wslink.py @@ -2,6 +2,7 @@ from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass from homeassistant.const import ( + CONCENTRATION_PARTS_PER_BILLION, DEGREE, PERCENTAGE, UV_INDEX, @@ -39,6 +40,7 @@ from .const import ( CHILL_INDEX, DAILY_RAIN, DEW_POINT, + HCHO, HEAT_INDEX, HOURLY_RAIN, INDOOR_BATTERY, @@ -50,7 +52,9 @@ from .const import ( OUTSIDE_TEMP, RAIN, SOLAR_RADIATION, + T9_BATTERY, UV, + VOC, WBGT_TEMP, WEEKLY_RAIN, WIND_AZIMUT, @@ -60,6 +64,7 @@ from .const import ( YEARLY_RAIN, UnitOfBat, UnitOfDir, + VOCLevel, ) from .sensors_common import WeatherSensorEntityDescription from .utils import battery_level, to_float, to_int, wind_dir_to_text @@ -518,4 +523,30 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = ( suggested_display_precision=2, value_fn=to_float, ), + 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 6eb5a49..bcd175f 100644 --- a/custom_components/sws12500/translations/cs.json +++ b/custom_components/sws12500/translations/cs.json @@ -308,6 +308,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 76ab3f5..a693980 100644 --- a/custom_components/sws12500/translations/en.json +++ b/custom_components/sws12500/translations/en.json @@ -315,6 +315,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 be52448..035be3b 100644 --- a/custom_components/sws12500/utils.py +++ b/custom_components/sws12500/utils.py @@ -24,15 +24,19 @@ from homeassistant.helpers.translation import async_get_translations from .const import ( AZIMUT, + CONNECTION_GATED_SENSORS, + DATABASE_PATH, DEV_DBG, OUTSIDE_HUMIDITY, OUTSIDE_TEMP, REMAP_ITEMS, REMAP_WSLINK_ITEMS, SENSORS_TO_LOAD, + VOC_LEVEL_MAP, WIND_SPEED, UnitOfBat, UnitOfDir, + VOCLevel, ) _LOGGER = logging.getLogger(__name__) @@ -52,9 +56,7 @@ async def translations( language: str = 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 None @@ -74,15 +76,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: str = cast("str", 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( @@ -93,17 +91,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: str, - update_value: str | list[str] | bool, -) -> 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 @@ -380,11 +371,7 @@ def chill_index( 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 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}