Michal
/
SWS-12500-custom-component
mirror of https://github.com/schizza/SWS-12500-custom-component
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
ecowitt_support
SchiZzA 2026-05-11 23:55:36 +02:00
parent 1d2c1b4be3
commit 21dfa61cd5
No known key found for this signature in database
6 changed files with 336 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
custom_components/sws12500/const.py
View File

@ -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."""

31
custom_components/sws12500/sensors_wslink.py
View File

@ -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,
),
)

16
custom_components/sws12500/translations/cs.json
View File

@ -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": {

16
custom_components/sws12500/translations/en.json
View File

@ -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": {

33
custom_components/sws12500/utils.py
View File

@ -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

231
tests/test_t9_air_quality.py Normal file
View File

@ -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}