Compare commits

...

2 Commits

Author SHA1 Message Date
SchiZzA 0e5a8ce10c
fix(wslink): compute heat and wind chill index when the station omits them
The WU/PWS platform derives both indices from temperature, humidity and wind
(value_from_data_fn=heat_index / chill_index), but the WSLink descriptions only
read the station's own t1heat / t1chill via value_fn=to_float. Stations that do
not report those fields still got both entities, because
_auto_enable_derived_sensors creates them as soon as the raw inputs arrive, so
they stayed Unavailable forever.

Add wslink_heat_index / wslink_chill_index: prefer the station's reading, compute
it otherwise. The conversions are the subtle part. The WSLink payload is metric
and both entities are declared in Celsius, while the NWS formulas are defined for
Fahrenheit, and chill_index converts the temperature but not the wind speed even
though its formula needs mph. So the temperature goes C -> F in, the wind m/s ->
mph, and the result F -> C back out. For the reported payload (6.2 C, 40 %,
30.6 m/s) that is -1.94 C; leaving the wind in m/s would have reported 0.34 C.

The native unit stays Celsius on purpose: switching it to Fahrenheit to match the
WU platform would misread the values of stations that do send t1heat / t1chill.

Also guard on missing temperature/humidity/wind before delegating, so a payload
without those fields no longer logs an error on every push.
2026-07-26 01:20:56 +02:00
SchiZzA 7c9343f90d
fix(sws12500): harden credential validation, URL fallback and push timeouts
- Reject blank or whitespace-only PWS credentials in both the config flow and
  the options flow. INVALID_CREDENTIALS only held placeholder strings, so an
  empty API ID or key passed validation, the entry was created, and
  _validate_credentials then rejected every incoming packet.
- Fall back to "UNKNOWN" host/port when get_url raises NoURLAvailableError.
  The values are shown as Ecowitt setup instructions only, so an unresolvable
  Home Assistant URL must not make the step unreachable.
- Catch TimeoutError alongside ClientError in the Windy and Pocasi push paths.
  A timeout is not a ClientError and used to escape into the webhook handler,
  answering the station with HTTP 500 even though the measured data was
  already stored.
- Add regression tests for all three and pin that 0-5 battery descriptions
  really reach the WSLink sensor platform.
2026-07-26 00:39:15 +02:00
10 changed files with 500 additions and 19 deletions

View File

@ -9,9 +9,9 @@ import voluptuous as vol
from yarl import URL
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import selector
from homeassistant.helpers.network import get_url
from homeassistant.helpers.network import NoURLAvailableError, get_url
from .conflicts import ERROR_MUTUALLY_EXCLUSIVE
from .const import (
@ -42,6 +42,31 @@ from .const import (
_PASSWORD_SELECTOR = selector.TextSelector(selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD))
def _is_invalid_credential(value: Any) -> bool:
"""Return True when a PWS/WSLink credential is unusable.
Blank (or whitespace-only) values are as useless as the placeholder strings in
`INVALID_CREDENTIALS`: `_validate_credentials` rejects every incoming packet if
either option is empty, so the entry would be created but never receive data.
"""
return not isinstance(value, str) or not value.strip() or value in INVALID_CREDENTIALS
def _ha_url_placeholders(hass: HomeAssistant) -> tuple[str, str]:
"""Return (host, port) of the Home Assistant URL for the Ecowitt instructions.
`get_url` raises `NoURLAvailableError` when HA cannot resolve any of its own
URLs. These values are shown as setup instructions only, so fall back to a
placeholder instead of letting the config flow abort.
"""
try:
url: URL = URL(get_url(hass))
except NoURLAvailableError:
return "UNKNOWN", "UNKNOWN"
return url.host or "UNKNOWN", str(url.port)
class ConfigOptionsFlowHandler(OptionsFlow):
"""Handle WeatherStation ConfigFlow."""
@ -160,9 +185,9 @@ class ConfigOptionsFlowHandler(OptionsFlow):
# legacy one while Ecowitt is active would corrupt those entities.
if self.ecowitt.get(ECOWITT_ENABLED):
errors["base"] = ERROR_MUTUALLY_EXCLUSIVE
elif user_input[API_ID] in INVALID_CREDENTIALS or user_input.get(API_ID, "") == "":
elif _is_invalid_credential(user_input.get(API_ID)):
errors[API_ID] = "valid_credentials_api"
elif user_input[API_KEY] in INVALID_CREDENTIALS or user_input.get(API_KEY, "") == "":
elif _is_invalid_credential(user_input.get(API_KEY)):
errors[API_KEY] = "valid_credentials_key"
elif user_input[API_KEY] == user_input[API_ID]:
errors["base"] = "valid_credentials_match"
@ -258,8 +283,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
else:
return self.async_create_entry(title=DOMAIN, data=self.retain_data(user_input))
url: URL = URL(get_url(self.hass))
host = url.host or "UNKNOWN"
host, port = _ha_url_placeholders(self.hass)
ecowitt_schema = {
vol.Required(
@ -277,7 +301,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
data_schema=vol.Schema(ecowitt_schema),
description_placeholders={
"url": host,
"port": str(url.port),
"port": port,
"webhook_id": webhook,
},
errors=errors,
@ -358,9 +382,9 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input is None:
return self.async_show_form(step_id="pws", data_schema=vol.Schema(self.pws_schema), errors=errors)
if user_input[API_ID] in INVALID_CREDENTIALS:
if _is_invalid_credential(user_input.get(API_ID)):
errors[API_ID] = "valid_credentials_api"
elif user_input[API_KEY] in INVALID_CREDENTIALS:
elif _is_invalid_credential(user_input.get(API_KEY)):
errors[API_KEY] = "valid_credentials_key"
elif user_input[API_KEY] == user_input[API_ID]:
errors["base"] = "valid_credentials_match"
@ -383,8 +407,7 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input is None:
webhook = secrets.token_hex(8)
url: URL = URL(get_url(self.hass))
host = url.host or "UNKNOWN"
host, port = _ha_url_placeholders(self.hass)
ecowitt_schema = {
vol.Required(ECOWITT_WEBHOOK_ID, default=webhook): str,
@ -396,7 +419,7 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
data_schema=vol.Schema(ecowitt_schema),
description_placeholders={
"url": host,
"port": str(url.port),
"port": port,
"webhook_id": webhook,
},
)

View File

@ -180,7 +180,10 @@ class PocasiPush:
_LOGGER.critical(POCASI_CZ_UNEXPECTED)
await self._disable_pocasi(POCASI_CZ_UNEXPECTED)
except ClientError as ex:
# TimeoutError is not a ClientError: an `async_timeout`/`asyncio` timeout would
# otherwise escape into the webhook handler and answer the station with HTTP 500,
# even though the measured data was already stored.
except (ClientError, TimeoutError) as ex:
self.last_status = "client_error"
# Store only the exception class - last_error is surfaced via entity
# attributes; str(ex) could embed the request URL.

View File

@ -69,7 +69,15 @@ from .const import (
VOCLevel,
)
from .sensors_common import WeatherSensorEntityDescription
from .utils import battery_level, to_float, to_int, voc_level_to_text, wind_dir_to_text
from .utils import (
battery_level,
to_float,
to_int,
voc_level_to_text,
wind_dir_to_text,
wslink_chill_index,
wslink_heat_index,
)
SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
WeatherSensorEntityDescription(
@ -457,6 +465,9 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
replacement_entity_key=CH8_BATTERY,
entity_registry_enabled_default=False,
),
# `value_from_data_fn` takes precedence in `WeatherSensor.native_value`, so these
# use the station's own t1heat/t1chill when present and compute them otherwise -
# stations that do not report them used to leave both entities Unavailable forever.
WeatherSensorEntityDescription(
key=HEAT_INDEX,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
@ -466,7 +477,7 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
suggested_display_precision=2,
icon="mdi:weather-sunny",
translation_key=HEAT_INDEX,
value_fn=to_float,
value_from_data_fn=wslink_heat_index,
),
WeatherSensorEntityDescription(
key=CHILL_INDEX,
@ -477,7 +488,7 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
suggested_display_precision=2,
icon="mdi:weather-sunny",
translation_key=CHILL_INDEX,
value_fn=to_float,
value_from_data_fn=wslink_chill_index,
),
WeatherSensorEntityDescription(
key=OUTSIDE_BATTERY,

View File

@ -15,7 +15,7 @@ from __future__ import annotations
import logging
import math
from typing import Any
from typing import Any, Final
from py_typecheck.core import checked_or
@ -26,8 +26,10 @@ from homeassistant.helpers.translation import async_get_translations
from .const import (
AZIMUT,
CHILL_INDEX,
CONNECTION_GATED_SENSORS,
DEV_DBG,
HEAT_INDEX,
OUTSIDE_HUMIDITY,
OUTSIDE_TEMP,
REMAP_ITEMS,
@ -392,3 +394,56 @@ def battery_5step_to_pct(value: Any) -> int | None:
return None
return round(min(max(step, 0), 5) / 5 * 100)
# The NWS wind-chill formula is defined for mph; WSLink reports wind in m/s.
_MS_TO_MPH: Final = 2.236936
def wslink_heat_index(data: dict[str, Any]) -> float | None:
"""Heat index for a WSLink payload, in Celsius.
Prefers the station's own `t1heat` reading. Not every WSLink station sends one,
and without a fallback the entity - which `_auto_enable_derived_sensors` creates
as soon as temperature and humidity arrive - stays Unavailable forever.
`heat_index` takes Celsius via `convert=True` but always returns Fahrenheit, so
the result is converted back to match this entity's native Celsius unit.
"""
if (reported := to_float(data.get(HEAT_INDEX))) is not None:
return reported
# Guard here rather than letting heat_index log an error per push: a payload
# without these fields is normal, not a fault.
if to_float(data.get(OUTSIDE_TEMP)) is None or to_float(data.get(OUTSIDE_HUMIDITY)) is None:
return None
value_f = heat_index(data, convert=True)
return None if value_f is None else round(fahrenheit_to_celsius(value_f), 2)
def wslink_chill_index(data: dict[str, Any]) -> float | None:
"""Wind chill for a WSLink payload, in Celsius.
Prefers the station's own `t1chill` reading; see `wslink_heat_index` for why a
fallback is needed.
`chill_index` converts the temperature but *not* the wind speed, and its formula
expects mph, so the m/s reading is converted here before the Fahrenheit result is
turned back into Celsius.
"""
if (reported := to_float(data.get(CHILL_INDEX))) is not None:
return reported
temp_c = to_float(data.get(OUTSIDE_TEMP))
wind_ms = to_float(data.get(WIND_SPEED))
if temp_c is None or wind_ms is None:
return None
value_f = chill_index(
{
OUTSIDE_TEMP: celsius_to_fahrenheit(temp_c),
WIND_SPEED: wind_ms * _MS_TO_MPH,
}
)
return None if value_f is None else round(fahrenheit_to_celsius(value_f), 2)

View File

@ -311,7 +311,10 @@ class WindyPush:
reason=f"Unable to send data to Windy ({WINDY_MAX_RETRIES} times). Disabling resend option for now. Please check your Windy configuration and enable this feature afterwards."
)
except ClientError as ex:
# TimeoutError is not a ClientError: an `async_timeout`/`asyncio` timeout would
# otherwise escape into the webhook handler and answer the station with HTTP 500,
# even though the measured data was already stored.
except (ClientError, TimeoutError) as ex:
self.last_status = "client_error"
# Store only the exception class - last_error is surfaced via entity
# attributes; str(ex) could embed the request URL.

View File

@ -81,13 +81,35 @@ def test_level_descriptions_generated_from_battery_non_binary() -> None:
def test_level_batteries_are_not_also_plain_sensors() -> None:
"""A 0-5 battery must appear exactly once in each platform's description set."""
"""A 0-5 battery must not be hand-defined on top of its generated description.
`BATTERY_LEVEL_SENSORS` is spliced into `SENSOR_TYPES_WSLINK`, so one occurrence
there is the generated one and is expected; `SENSOR_TYPES_WEATHER_API` does not
splice them in, so zero is expected there. "At most once" is therefore the
invariant that holds for both - a second entry is a duplicate entity.
Complete *absence* of a description is caught by
`test_level_descriptions_generated_from_battery_non_binary`, which pins the
generated set to `BATTERY_NON_BINARY` exactly.
"""
for sensor_types in (SENSOR_TYPES_WSLINK, SENSOR_TYPES_WEATHER_API):
keys = [desc.key for desc in sensor_types]
for key in BATTERY_NON_BINARY:
assert keys.count(key) <= 1, f"{key} defined more than once"
def test_level_batteries_reach_the_sensor_platform() -> None:
"""Every 0-5 battery must actually be spliced into the WSLink platform set.
The generated tuple existing is not enough - if the `*BATTERY_LEVEL_SENSORS`
splice were dropped, the classification tests would still pass while the
percentage entities silently disappeared.
"""
keys = [desc.key for desc in SENSOR_TYPES_WSLINK]
for key in BATTERY_NON_BINARY:
assert key in keys, f"{key} is classified as a 0-5 battery but has no sensor description"
def test_binary_batteries_are_not_plain_sensors_too() -> None:
"""0/1 batteries belong to the binary platform only.

View File

@ -29,6 +29,7 @@ from custom_components.sws12500.const import (
WSLINK_ADDON_PORT,
)
from homeassistant import config_entries
from homeassistant.helpers.network import NoURLAvailableError
@pytest.mark.asyncio
@ -127,6 +128,81 @@ async def test_config_flow_user_invalid_credentials_api_key(
assert result2["errors"][API_KEY] == "valid_credentials_key"
@pytest.mark.parametrize("blank", ["", " "])
@pytest.mark.parametrize("field", [API_ID, API_KEY])
@pytest.mark.asyncio
async def test_config_flow_user_rejects_blank_credentials(
hass, enable_custom_integrations, field, blank
) -> None:
"""Blank PWS credentials must not create an entry.
`INVALID_CREDENTIALS` holds placeholder strings only, so an empty (or
whitespace-only) value used to pass validation. The entry was created but
`_validate_credentials` then rejected every incoming packet, leaving the
integration permanently without data.
"""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
form = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "pws"}
)
assert form["step_id"] == "pws"
user_input = {
API_ID: "ok_id",
API_KEY: "ok_key",
WSLINK: False,
DEV_DBG: False,
}
user_input[field] = blank
result2 = await hass.config_entries.flow.async_configure(
form["flow_id"], user_input=user_input
)
assert result2["type"] == "form"
assert result2["step_id"] == "pws"
expected = "valid_credentials_api" if field is API_ID else "valid_credentials_key"
assert result2["errors"][field] == expected
@pytest.mark.parametrize("blank", ["", " "])
@pytest.mark.parametrize("field", [API_ID, API_KEY])
@pytest.mark.asyncio
async def test_options_flow_basic_rejects_blank_credentials(
hass, enable_custom_integrations, field, blank
) -> None:
"""Same blank-credential rule applies when the legacy endpoint is re-enabled."""
entry = MockConfigEntry(
domain=DOMAIN,
data={},
options={LEGACY_ENABLED: False, ECOWITT_ENABLED: False},
)
entry.add_to_hass(hass)
init = await hass.config_entries.options.async_init(entry.entry_id)
form = await hass.config_entries.options.async_configure(
init["flow_id"], user_input={"next_step_id": "basic"}
)
assert form["step_id"] == "basic"
user_input = {
API_ID: "ok_id",
API_KEY: "ok_key",
WSLINK: False,
DEV_DBG: False,
LEGACY_ENABLED: True,
}
user_input[field] = blank
result = await hass.config_entries.options.async_configure(
init["flow_id"], user_input=user_input
)
assert result["type"] == "form"
expected = "valid_credentials_api" if field is API_ID else "valid_credentials_key"
assert result["errors"][field] == expected
@pytest.mark.asyncio
async def test_config_flow_user_invalid_credentials_match(
hass, enable_custom_integrations
@ -410,6 +486,70 @@ async def test_options_flow_ecowitt_uses_get_url_placeholders_and_webhook_defaul
assert done["data"][ECOWITT_ENABLED] is True
@pytest.mark.asyncio
async def test_options_flow_ecowitt_survives_no_url_available(
hass, enable_custom_integrations
) -> None:
"""`get_url` raising must not abort the Ecowitt options step.
The host/port are shown as setup instructions only. HA can fail to resolve any
of its own URLs (no internal/external URL configured), and letting that escape
made the Ecowitt setup unreachable.
"""
entry = MockConfigEntry(
domain=DOMAIN,
data={},
options={ECOWITT_WEBHOOK_ID: "", ECOWITT_ENABLED: False, LEGACY_ENABLED: False},
)
entry.add_to_hass(hass)
init = await hass.config_entries.options.async_init(entry.entry_id)
with patch(
"custom_components.sws12500.config_flow.get_url",
side_effect=NoURLAvailableError,
):
form = await hass.config_entries.options.async_configure(
init["flow_id"], user_input={"next_step_id": "ecowitt"}
)
assert form["type"] == "form"
placeholders = form.get("description_placeholders") or {}
assert placeholders["url"] == "UNKNOWN"
assert placeholders["port"] == "UNKNOWN"
assert placeholders["webhook_id"]
@pytest.mark.asyncio
async def test_config_flow_ecowitt_survives_no_url_available(
hass, enable_custom_integrations
) -> None:
"""The initial Ecowitt step stays usable when HA cannot resolve its own URL."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"custom_components.sws12500.config_flow.get_url",
side_effect=NoURLAvailableError,
):
form = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"next_step_id": "ecowitt"}
)
assert form["type"] == "form"
placeholders = form.get("description_placeholders") or {}
assert placeholders["url"] == "UNKNOWN"
assert placeholders["port"] == "UNKNOWN"
# The step is still completable without a resolvable URL.
done = await hass.config_entries.flow.async_configure(
form["flow_id"],
user_input={
ECOWITT_WEBHOOK_ID: placeholders["webhook_id"],
ECOWITT_ENABLED: True,
},
)
assert done["type"] == "create_entry"
assert done["data"][ECOWITT_ENABLED] is True
@pytest.mark.asyncio
async def test_options_flow_wslink_port_setup(hass, enable_custom_integrations) -> None:
"""The WSLink add-on port step shows a form and stores the port."""

View File

@ -338,6 +338,40 @@ async def test_push_data_to_server_client_error_increments_and_disables_after_th
)
@pytest.mark.asyncio
async def test_push_data_to_server_timeout_is_handled_like_client_error(
monkeypatch, hass
):
"""A network timeout must not escape into the webhook handler.
`TimeoutError` is not a subclass of `ClientError`, so it used to propagate out
of `push_data_to_server`, through the awaiting coordinator, and answer the
station with HTTP 500 - even though the measured data was already stored.
"""
entry = _make_entry()
pp = PocasiPush(hass, entry)
monkeypatch.setattr(
"custom_components.sws12500.pocasti_cz.update_options",
_write_through_update_options(entry),
)
monkeypatch.setattr("custom_components.sws12500.pocasti_cz._LOGGER.critical", MagicMock())
session = _FakeSession(exc=TimeoutError("timed out"))
monkeypatch.setattr(
"custom_components.sws12500.pocasti_cz.async_get_clientsession",
lambda _h: session,
)
pp.next_update = dt_util.utcnow() - timedelta(seconds=1)
await pp.push_data_to_server({"x": 1}, "WU")
assert pp.last_status == "client_error"
assert pp.last_error == "TimeoutError"
assert pp.invalid_response_count == 1
assert pp.enabled is True
def test_verify_response_logs_debug_when_logger_enabled(monkeypatch, hass):
entry = _make_entry(logger=True)
pp = PocasiPush(hass, entry)

View File

@ -449,6 +449,43 @@ async def test_push_data_to_windy_client_error_increments_and_disables_after_thr
)
@pytest.mark.asyncio
async def test_push_data_to_windy_timeout_is_handled_like_client_error(
monkeypatch, hass
):
"""A network timeout must not escape into the webhook handler.
`TimeoutError` is not a subclass of `ClientError`, so it used to propagate out
of `push_data_to_windy`, through the awaiting coordinator, and answer the
station with HTTP 500 - even though the measured data was already stored.
"""
entry = _make_entry()
wp = WindyPush(hass, entry)
wp.next_update = dt_util.utcnow() - timedelta(seconds=1)
monkeypatch.setattr(
"custom_components.sws12500.windy_func.update_options", AsyncMock(return_value=True)
)
monkeypatch.setattr("custom_components.sws12500.windy_func._LOGGER.critical", MagicMock())
monkeypatch.setattr(
"custom_components.sws12500.windy_func.persistent_notification.async_create",
MagicMock(),
)
session = _FakeSession(exc=TimeoutError("timed out"))
monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session,
)
ok = await wp.push_data_to_windy({"a": "b"})
assert ok is True
assert wp.last_status == "client_error"
assert wp.last_error == "TimeoutError"
assert wp.invalid_response_count == 1
@pytest.mark.asyncio
async def test_push_data_to_windy_client_error_disable_failure_logs_debug(
monkeypatch, hass

View File

@ -0,0 +1,153 @@
"""Heat / wind-chill indices must work on WSLink too.
`_auto_enable_derived_sensors` creates both entities as soon as temperature and
humidity (resp. wind speed) arrive, but the WSLink descriptions only ever read the
station's own ``t1heat`` / ``t1chill``. A station that does not report those left both
entities Unavailable forever, while the WU/PWS platform computed them.
The payload is metric, the NWS formulas are Fahrenheit-based (and mph for wind chill),
and the WSLink entities are declared in Celsius - so the conversions matter as much as
the fallback itself.
"""
from __future__ import annotations
import pytest
from custom_components.sws12500.const import (
CHILL_INDEX,
HEAT_INDEX,
OUTSIDE_HUMIDITY,
OUTSIDE_TEMP,
WIND_SPEED,
)
from custom_components.sws12500.sensors_wslink import SENSOR_TYPES_WSLINK
from custom_components.sws12500.utils import (
celsius_to_fahrenheit,
chill_index,
fahrenheit_to_celsius,
heat_index,
wslink_chill_index,
wslink_heat_index,
)
# The remapped form of a real payload from a station that sends neither t1heat nor
# t1chill (t1tem/t1hum/t1ws in metric units).
LIVE_PAYLOAD: dict[str, str] = {
OUTSIDE_TEMP: "6.2",
OUTSIDE_HUMIDITY: "40",
WIND_SPEED: "30.6",
}
def _description(key: str):
return next(desc for desc in SENSOR_TYPES_WSLINK if desc.key == key)
# ---------------------------------------------------------------------------
# Fallback
# ---------------------------------------------------------------------------
def test_indices_are_computed_when_the_station_omits_them() -> None:
"""The reported bug: both entities stayed empty for this exact payload."""
assert wslink_heat_index(LIVE_PAYLOAD) is not None
assert wslink_chill_index(LIVE_PAYLOAD) is not None
def test_computed_values_are_celsius_and_plausible() -> None:
"""6.2 C at 40% RH with strong wind: heat index near ambient, chill below it."""
heat = wslink_heat_index(LIVE_PAYLOAD)
chill = wslink_chill_index(LIVE_PAYLOAD)
assert heat is not None and chill is not None
# Sanity-check the magnitude - a Fahrenheit result leaking through would be ~39/28.
assert -10 < heat < 15, heat
assert -10 < chill < 15, chill
# Wind chill must be colder than the still-air temperature at 30.6 m/s.
assert chill < 6.2
def test_wind_is_converted_to_mph() -> None:
"""`chill_index` converts temperature but not wind, and its formula needs mph.
Feeding m/s straight in understates the wind and yields a warmer chill.
"""
correct = wslink_chill_index(LIVE_PAYLOAD)
naive_f = chill_index(
{OUTSIDE_TEMP: celsius_to_fahrenheit(6.2), WIND_SPEED: 30.6}, # m/s, unconverted
)
assert naive_f is not None
naive = round(fahrenheit_to_celsius(naive_f), 2)
assert correct is not None
assert correct < naive, f"unconverted wind gives {naive}, converted gives {correct}"
# ---------------------------------------------------------------------------
# The station's own reading still wins
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("fn", "key"),
[(wslink_heat_index, HEAT_INDEX), (wslink_chill_index, CHILL_INDEX)],
ids=["heat", "chill"],
)
def test_station_reported_value_takes_precedence(fn, key) -> None:
"""t1heat / t1chill are already Celsius and must be passed through untouched."""
assert fn({**LIVE_PAYLOAD, key: "3.5"}) == 3.5
@pytest.mark.parametrize(
("fn", "key"),
[(wslink_heat_index, HEAT_INDEX), (wslink_chill_index, CHILL_INDEX)],
ids=["heat", "chill"],
)
def test_empty_reported_value_falls_back_to_computing(fn, key) -> None:
"""The station sends "" for absent fields; that must not shadow the fallback."""
assert fn({**LIVE_PAYLOAD, key: ""}) is not None
# ---------------------------------------------------------------------------
# Missing inputs stay quiet
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("missing", [OUTSIDE_TEMP, OUTSIDE_HUMIDITY])
def test_heat_index_needs_temp_and_humidity(missing) -> None:
payload = {k: v for k, v in LIVE_PAYLOAD.items() if k != missing}
assert wslink_heat_index(payload) is None
@pytest.mark.parametrize("missing", [OUTSIDE_TEMP, WIND_SPEED])
def test_chill_index_needs_temp_and_wind(missing) -> None:
payload = {k: v for k, v in LIVE_PAYLOAD.items() if k != missing}
assert wslink_chill_index(payload) is None
def test_no_inputs_at_all_is_none() -> None:
assert wslink_heat_index({}) is None
assert wslink_chill_index({}) is None
# ---------------------------------------------------------------------------
# Wiring
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(("key", "expected"), [(HEAT_INDEX, wslink_heat_index), (CHILL_INDEX, wslink_chill_index)])
def test_descriptions_use_the_wslink_helpers(key, expected) -> None:
"""`value_from_data_fn` wins in WeatherSensor.native_value, so it must be set."""
desc = _description(key)
assert desc.value_from_data_fn is expected
def test_wu_platform_still_computes_in_fahrenheit() -> None:
"""The WU descriptions are Fahrenheit-native; they must keep the raw helpers."""
from custom_components.sws12500.sensors_weather import SENSOR_TYPES_WEATHER_API
by_key = {d.key: d for d in SENSOR_TYPES_WEATHER_API}
assert by_key[HEAT_INDEX].value_from_data_fn is heat_index
assert by_key[CHILL_INDEX].value_from_data_fn is chill_index