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
SWS-12500-custom-component/custom_components/sws12500/utils.py

503 lines
14 KiB
Python
Raw Blame History

"""Utils for SWS12500.
This module contains small helpers used across the integration.
Notable responsibilities:
- Payload remapping: convert raw station/webhook field names into stable internal keys.
- Auto-discovery helpers: detect new payload fields that are not enabled yet and persist them
to config entry options so sensors can be created dynamically.
- Formatting/conversion helpers (wind direction text, battery mapping, temperature conversions).
Keeping these concerns in one place avoids duplicating logic in the webhook handler and entity code.
"""
from __future__ import annotations
import logging
import math
from typing import Any
from py_typecheck.core import checked_or
from homeassistant.components import persistent_notification
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
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__)
async def translations(
hass: HomeAssistant,
translation_domain: str,
translation_key: str,
*,
key: str = "message",
category: str = "notify",
) -> str | None:
"""Get translated keys for domain."""
localize_key = f"component.{translation_domain}.{category}.{translation_key}.{key}"
language: str = hass.config.language
_translations = await async_get_translations(hass, language, category, [translation_domain])
if localize_key in _translations:
return _translations[localize_key]
return None
async def translated_notification(
hass: HomeAssistant,
translation_domain: str,
translation_key: str,
translation_placeholders: dict[str, str] | None = None,
notification_id: str | None = None,
*,
key: str = "message",
category: str = "notify",
):
"""Translate notification."""
localize_key = f"component.{translation_domain}.{category}.{translation_key}.{key}"
localize_title = f"component.{translation_domain}.{category}.{translation_key}.title"
language: str = hass.config.language
_translations = await async_get_translations(hass, language, category, [translation_domain])
if localize_key in _translations:
if not translation_placeholders:
persistent_notification.async_create(
hass,
_translations[localize_key],
_translations[localize_title],
notification_id,
)
else:
message = _translations[localize_key].format(**translation_placeholders)
persistent_notification.async_create(hass, message, _translations[localize_title], notification_id)
async def update_options(hass: HomeAssistant, entry: ConfigEntry, update_key: str, update_value: Any) -> bool:
"""Update config.options entry."""
conf = {**entry.options}
conf[update_key] = update_value
return hass.config_entries.async_update_entry(entry, options=conf)
def anonymize(
data: dict[str, str | int | float | bool],
) -> dict[str, str | int | float | bool]:
"""Anonymize received data for safe logging.
- Keep all keys, but mask sensitive values.
- Do not raise on unexpected/missing keys.
"""
secrets = {"ID", "PASSWORD", "wsid", "wspw"}
return {k: ("***" if k in secrets else v) for k, v in data.items()}
def remap_items(entities: dict[str, str]) -> dict[str, str]:
"""Remap legacy (WU-style) payload field names into internal sensor keys.
The station sends short/legacy field names (e.g. "tempf", "humidity"). Internally we use
stable keys from `const.py` (e.g. "outside_temp", "outside_humidity"). This function produces
a normalized dict that the rest of the integration can work with.
"""
return {REMAP_ITEMS[key]: value for key, value in entities.items() if key in REMAP_ITEMS}
def remap_wslink_items(entities: dict[str, str]) -> dict[str, str]:
"""Remap items in query for WSLink API."""
items: dict[str, str] = {}
for item, value in entities.items():
if item in REMAP_WSLINK_ITEMS:
items[REMAP_WSLINK_ITEMS[item]] = value
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
def loaded_sensors(config_entry: ConfigEntry) -> list[str]:
"""Return sensor keys currently enabled for this config entry.
Auto-discovery persists new keys into `config_entry.options[SENSORS_TO_LOAD]`. The sensor
platform uses this list to decide which entities to create.
"""
return config_entry.options.get(SENSORS_TO_LOAD) or []
def check_disabled(items: dict[str, str], config_entry: ConfigEntry) -> list[str] | None:
"""Detect payload fields that are not enabled yet (auto-discovery).
The integration supports "auto-discovery" of sensors: when the station starts sending a new
field, we can automatically enable and create the corresponding entity.
This helper compares the normalized payload keys (`items`) with the currently enabled sensor
keys stored in options (`SENSORS_TO_LOAD`) and returns the missing keys.
Returns:
- list[str] of newly discovered sensor keys (to be added/enabled), or
- None if no new keys were found.
Notes:
- Logging is controlled via `DEV_DBG` because payloads can arrive frequently.
"""
log = checked_or(config_entry.options.get(DEV_DBG), bool, False)
entityFound: bool = False
_loaded_sensors: list[str] = loaded_sensors(config_entry)
missing_sensors: list[str] = []
for item in items:
if log:
_LOGGER.info("Checking %s", item)
if item not in _loaded_sensors:
missing_sensors.append(item)
entityFound = True
if log:
_LOGGER.info("Add sensor (%s) to loading queue", item)
return missing_sensors if entityFound else None
def wind_dir_to_text(deg: float | str | None) -> UnitOfDir | None:
"""Return wind direction in text representation.
Accepts the raw payload value (str), a float, or None. A direction of 0 - or
a missing/invalid value - is treated as "no reading" (calm) and returns None,
so a missing wind direction does not render as North.
Returns UnitOfDir or None
"""
_deg = to_float(deg)
if _deg is None or _deg == 0:
return None
azimut = AZIMUT[int(abs((_deg - 11.25) % 360) / 22.5)]
_LOGGER.debug("wind_dir: %s", azimut)
return azimut
def battery_level(battery: int | str | None) -> UnitOfBat:
"""Return battery level.
WSLink payload values often arrive as strings (e.g. "0"/"1"), so we accept
both ints and strings and coerce to int before mapping.
Returns UnitOfBat
"""
level_map: dict[int, UnitOfBat] = {
0: UnitOfBat.LOW,
1: UnitOfBat.NORMAL,
}
if (battery is None) or (battery == ""):
return UnitOfBat.UNKNOWN
vi: int
if isinstance(battery, int):
vi = battery
else:
try:
vi = int(battery)
except ValueError:
return UnitOfBat.UNKNOWN
return level_map.get(vi, UnitOfBat.UNKNOWN)
def battery_level_to_icon(battery: UnitOfBat) -> str:
"""Return battery level in icon representation.
Returns str
"""
icons = {
UnitOfBat.LOW: "mdi:battery-low",
UnitOfBat.NORMAL: "mdi:battery",
}
return icons.get(battery, "mdi:battery-unknown")
def fahrenheit_to_celsius(fahrenheit: float) -> float:
"""Convert Fahrenheit to Celsius."""
return (fahrenheit - 32) * 5.0 / 9.0
def celsius_to_fahrenheit(celsius: float) -> float:
"""Convert Celsius to Fahrenheit."""
return celsius * 9.0 / 5.0 + 32
def to_int(val: Any) -> int | None:
"""Convert int or string to int."""
if val is None:
return None
if isinstance(val, str) and val.strip() == "":
return None
try:
v = int(val)
except (TypeError, ValueError):
return None
else:
return v
def to_float(val: Any) -> float | None:
"""Convert int or string to float."""
if val is None:
return None
if isinstance(val, str) and val.strip() == "":
return None
try:
v = float(val)
except (TypeError, ValueError):
return None
else:
return v
def heat_index(data: dict[str, int | float | str], convert: bool = False) -> float | None:
"""Calculate heat index from temperature.
data: dict with temperature and humidity
convert: bool, convert received data from Celsius to Fahrenheit
"""
if (temp := to_float(data.get(OUTSIDE_TEMP))) is None:
_LOGGER.error(
"We are missing/invalid OUTSIDE TEMP (%s), cannot calculate wind chill index.",
temp,
)
return None
if (rh := to_float(data.get(OUTSIDE_HUMIDITY))) is None:
_LOGGER.error(
"We are missing/invalid OUTSIDE HUMIDITY (%s), cannot calculate wind chill index.",
rh,
)
return None
adjustment = None
if convert:
temp = celsius_to_fahrenheit(temp)
simple = 0.5 * (temp + 61.0 + ((temp - 68.0) * 1.2) + (rh * 0.094))
if ((simple + temp) / 2) > 80:
full_index = (
-42.379
+ 2.04901523 * temp
+ 10.14333127 * rh
- 0.22475541 * temp * rh
- 0.00683783 * temp * temp
- 0.05481717 * rh * rh
+ 0.00122874 * temp * temp * rh
+ 0.00085282 * temp * rh * rh
- 0.00000199 * temp * temp * rh * rh
)
if rh < 13 and (80 <= temp <= 112):
adjustment = ((13 - rh) / 4) * math.sqrt((17 - abs(temp - 95)) / 17)
if rh > 80 and (80 <= temp <= 87):
adjustment = ((rh - 85) / 10) * ((87 - temp) / 5)
return round((full_index + adjustment if adjustment else full_index), 2)
return simple
def chill_index(data: dict[str, str | float | int], convert: bool = False) -> float | None:
"""Calculate wind chill index from temperature and wind speed.
data: dict with temperature and wind speed
convert: bool, convert received data from Celsius to Fahrenheit
"""
temp = to_float(data.get(OUTSIDE_TEMP))
wind = to_float(data.get(WIND_SPEED))
if temp is None:
_LOGGER.error(
"We are missing/invalid OUTSIDE TEMP (%s), cannot calculate wind chill index.",
temp,
)
return None
if wind is None:
_LOGGER.error(
"We are missing/invalid WIND SPEED (%s), cannot calculate wind chill index.",
wind,
)
return None
if convert:
temp = celsius_to_fahrenheit(temp)
return (
round(
((35.7 + (0.6215 * temp)) - (35.75 * (wind**0.16)) + (0.4275 * (temp * (wind**0.16)))),
2,
)
if temp < 50 and wind > 3
else temp
)
def voc_level_to_text(value: str | None) -> 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 = []
# if not Path(DATABASE_PATH).exists():
# _LOGGER.error("Database file not found: %s", DATABASE_PATH)
# return False
#
# conn = sqlite3.connect(DATABASE_PATH)
# db = conn.cursor()
#
# try:
# db.execute(
# """
# SELECT statistic_id, unit_of_measurement from statistics_meta
# WHERE statistic_id LIKE 'sensor.weather_station_sws%'
# """
# )
# rows = db.fetchall()
# 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)
# finally:
# conn.close()
#
# return sensor_units
#
#
# async def migrate_data(hass: HomeAssistant, sensor_id: str | None = None) -> int | bool:
# """Migrate data from mm/d to mm."""
#
# _LOGGER.debug("Sensor %s is required for data migration", sensor_id)
# updated_rows = 0
#
# if not Path(DATABASE_PATH).exists():
# _LOGGER.error("Database file not found: %s", DATABASE_PATH)
# return False
#
# conn = sqlite3.connect(DATABASE_PATH)
# db = conn.cursor()
#
# try:
# _LOGGER.info(sensor_id)
# db.execute(
# """
# UPDATE statistics_meta
# SET unit_of_measurement = 'mm'
# WHERE statistic_id = ?
# AND unit_of_measurement = 'mm/d';
# """,
# (sensor_id,),
# )
# updated_rows = db.rowcount
# conn.commit()
# _LOGGER.info(
# "Data migration completed successfully. Updated rows: %s for %s",
# updated_rows,
# sensor_id,
# )
#
# except sqlite3.Error as e:
# _LOGGER.error("Error during data migration: %s", e)
# finally:
# conn.close()
# return updated_rows
#
#
# def migrate_data_old(sensor_id: str | None = None):
# """Migrate data from mm/d to mm."""
# updated_rows = 0
#
# if not Path(DATABASE_PATH).exists():
# _LOGGER.error("Database file not found: %s", DATABASE_PATH)
# return False
#
# conn = sqlite3.connect(DATABASE_PATH)
# db = conn.cursor()
#
# try:
# _LOGGER.info(sensor_id)
# db.execute(
# """
# UPDATE statistics_meta
# SET unit_of_measurement = 'mm'
# WHERE statistic_id = ?
# AND unit_of_measurement = 'mm/d';
# """,
# (sensor_id,),
# )
# updated_rows = db.rowcount
# conn.commit()
# _LOGGER.info(
# "Data migration completed successfully. Updated rows: %s for %s",
# updated_rows,
# sensor_id,
# )
#
# except sqlite3.Error as e:
# _LOGGER.error("Error during data migration: %s", e)
# finally:
# conn.close()
# return updated_rows
Reference in New Issue View Git Blame Copy Permalink