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

Merge pull request #83 from schizza/recactor/fixes 

Recactor/fixes
fix/index_computing
Lukas Svoboda 2025-11-17 00:29:50 +01:00 committed by GitHub
parent 826a9a71cc 44d0ee5c7b
commit 6edaec73d8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 71 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
custom_components/sws12500/__init__.py
View File

@ -2,7 +2,7 @@
import logging
import aiohttp
import aiohttp.web
from aiohttp.web_exceptions import HTTPUnauthorized
from homeassistant.config_entries import ConfigEntry
@ -140,11 +140,11 @@ def register_path(
hass_data = hass.data.setdefault(DOMAIN, {})
debug = config.options.get(DEV_DBG)
_wslink = config.options.get(WSLINK)
_wslink = config.options.get(WSLINK, False)
routes: Routes = hass_data.get("routes") if "routes" in hass_data else None
routes: Routes = hass_data.get("routes", Routes())
if routes is None:
if not routes.routes:
routes = Routes()
_LOGGER.info("Routes not found, creating new routes")

9
custom_components/sws12500/routes.py
View File

@ -1,5 +1,6 @@
"""Store routes info."""
from collections.abc import Callable
from dataclasses import dataclass
from logging import getLogger
@ -14,7 +15,7 @@ class Route:
url_path: str
route: AbstractRoute
handler: callable
handler: Callable
enabled: bool = False
def __str__(self):
@ -29,7 +30,7 @@ class Routes:
"""Initialize routes."""
self.routes = {}
def switch_route(self, coordinator: callable, url_path: str):
def switch_route(self, coordinator: Callable, url_path: str):
"""Switch route."""
for url, route in self.routes.items():
@ -47,7 +48,7 @@ class Routes:
self,
url_path: str,
route: AbstractRoute,
handler: callable,
handler: Callable,
enabled: bool = False,
):
"""Add route."""
@ -55,7 +56,7 @@ class Routes:
def get_route(self, url_path: str) -> Route:
"""Get route."""
return self.routes.get(url_path)
return self.routes.get(url_path, Route)
def get_enabled(self) -> str:
"""Get enabled routes."""

47
custom_components/sws12500/sensor.py
View File

@ -2,7 +2,7 @@
import logging
from homeassistant.components.sensor import RestoreSensor, SensorEntity
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceEntryType
@ -12,10 +12,10 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import WeatherDataUpdateCoordinator
from .const import (
BATTERY_LIST,
CHILL_INDEX,
DOMAIN,
HEAT_INDEX,
OUTSIDE_BATTERY,
OUTSIDE_HUMIDITY,
OUTSIDE_TEMP,
SENSORS_TO_LOAD,
@ -23,12 +23,12 @@ from .const import (
WIND_DIR,
WIND_SPEED,
WSLINK,
BATTERY_LIST,
UnitOfBat,
)
from .sensors_common import WeatherSensorEntityDescription
from .sensors_weather import SENSOR_TYPES_WEATHER_API
from .sensors_wslink import SENSOR_TYPES_WSLINK
from .utils import chill_index, heat_index, battery_level_to_icon, battery_level_to_text
from .utils import battery_level_to_icon, battery_level_to_text, chill_index, heat_index
_LOGGER = logging.getLogger(__name__)
@ -44,12 +44,12 @@ async def async_setup_entry(
sensors_to_load: list = []
sensors: list = []
_wslink = config_entry.data.get(WSLINK)
_wslink = config_entry.options.get(WSLINK)
SENSOR_TYPES = SENSOR_TYPES_WSLINK if _wslink else SENSOR_TYPES_WEATHER_API
# Check if we have some sensors to load.
if sensors_to_load := config_entry.options.get(SENSORS_TO_LOAD):
if sensors_to_load := config_entry.options.get(SENSORS_TO_LOAD, []):
if WIND_DIR in sensors_to_load:
sensors_to_load.append(WIND_AZIMUT)
if (OUTSIDE_HUMIDITY in sensors_to_load) and (OUTSIDE_TEMP in sensors_to_load):
@ -65,9 +65,9 @@ async def async_setup_entry(
async_add_entities(sensors)
class WeatherSensor(
CoordinatorEntity[WeatherDataUpdateCoordinator], RestoreSensor, SensorEntity
):
class WeatherSensor( # pyright: ignore[reportIncompatibleVariableOverride]
CoordinatorEntity[WeatherDataUpdateCoordinator], SensorEntity
): # pyright: ignore[reportIncompatibleVariableOverride]
"""Implementation of Weather Sensor entity."""
_attr_has_entity_name = True
@ -94,12 +94,6 @@ class WeatherSensor(
self.coordinator.async_add_listener(self._handle_coordinator_update)
# prev_state_data = await self.async_get_last_sensor_data()
# prev_state = await self.async_get_last_state()
# if not prev_state:
# return
# self._data = prev_state_data.native_value
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
@ -110,30 +104,30 @@ class WeatherSensor(
self.async_write_ha_state()
@property
def native_value(self) -> str | int | float | None:
def native_value(self): # pyright: ignore[reportIncompatibleVariableOverride]
"""Return value of entity."""
_wslink = self.coordinator.config.options.get(WSLINK)
if self.coordinator.data and (WIND_AZIMUT in self.entity_description.key):
return self.entity_description.value_fn(self.coordinator.data.get(WIND_DIR))
return self.entity_description.value_fn(self.coordinator.data.get(WIND_DIR)) # pyright: ignore[ reportAttributeAccessIssue]
if (
self.coordinator.data
and (HEAT_INDEX in self.entity_description.key)
and not _wslink
):
return self.entity_description.value_fn(heat_index(self.coordinator.data))
return self.entity_description.value_fn(heat_index(self.coordinator.data)) # pyright: ignore[ reportAttributeAccessIssue]
if (
self.coordinator.data
and (CHILL_INDEX in self.entity_description.key)
and not _wslink
):
return self.entity_description.value_fn(chill_index(self.coordinator.data))
return self.entity_description.value_fn(chill_index(self.coordinator.data)) # pyright: ignore[ reportAttributeAccessIssue]
return (
None if self._data == "" else self.entity_description.value_fn(self._data)
None if self._data == "" else self.entity_description.value_fn(self._data) # pyright: ignore[ reportAttributeAccessIssue]
)
@property
@ -142,19 +136,20 @@ class WeatherSensor(
return generate_entity_id("sensor.{}", self.entity_description.key)
@property
def icon(self) -> str | None:
def icon(self) -> str | None: # pyright: ignore[reportIncompatibleVariableOverride]
"""Return the dynamic icon for battery representation."""
if self.entity_description.key in BATTERY_LIST:
try:
return battery_level_to_icon(self.native_value)
except Exception:
return "mdi:battery-unknown"
if self.native_value:
battery_level = battery_level_to_text(self.native_value)
return battery_level_to_icon(battery_level)
return battery_level_to_icon(UnitOfBat.UNKNOWN)
return self.entity_description.icon
@property
def device_info(self) -> DeviceInfo:
def device_info(self) -> DeviceInfo: # pyright: ignore[reportIncompatibleVariableOverride]
"""Device info."""
return DeviceInfo(
connections=set(),

20
custom_components/sws12500/sensors_wslink.py
View File

@ -17,14 +17,11 @@ from homeassistant.const import (
from .const import (
BARO_PRESSURE,
CH2_BATTERY,
CH2_HUMIDITY,
CH2_TEMP,
CH2_BATTERY,
INDOOR_BATTERY,
CH3_HUMIDITY,
CH3_TEMP,
CH4_HUMIDITY,
CH4_TEMP,
CHILL_INDEX,
DAILY_RAIN,
DEW_POINT,
@ -33,7 +30,6 @@ from .const import (
INDOOR_BATTERY,
INDOOR_HUMIDITY,
INDOOR_TEMP,
INDOOR_BATTERY,
MONTHLY_RAIN,
OUTSIDE_BATTERY,
OUTSIDE_HUMIDITY,
@ -41,6 +37,7 @@ from .const import (
RAIN,
SOLAR_RADIATION,
UV,
WBGT_TEMP,
WEEKLY_RAIN,
WIND_AZIMUT,
WIND_DIR,
@ -48,10 +45,9 @@ from .const import (
WIND_SPEED,
YEARLY_RAIN,
UnitOfDir,
WBGT_TEMP,
)
from .sensors_common import WeatherSensorEntityDescription
from .utils import battery_level_to_icon, battery_level_to_text, wind_dir_to_text
from .utils import wind_dir_to_text
SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
WeatherSensorEntityDescription(
@ -258,7 +254,7 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
suggested_unit_of_measurement=UnitOfTemperature.CELSIUS,
icon="mdi:weather-sunny",
translation_key=CH3_TEMP,
value_fn=lambda data: cast(float, data),
value_fn=lambda data: cast("float", data),
),
WeatherSensorEntityDescription(
key=CH3_HUMIDITY,
@ -267,7 +263,7 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.HUMIDITY,
icon="mdi:weather-sunny",
translation_key=CH3_HUMIDITY,
value_fn=lambda data: cast(int, data),
value_fn=lambda data: cast("int", data),
),
# WeatherSensorEntityDescription(
# key=CH4_TEMP,
@ -315,21 +311,21 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
translation_key=OUTSIDE_BATTERY,
icon="mdi:battery-unknown",
device_class=SensorDeviceClass.ENUM,
value_fn=lambda data: battery_level_to_text(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: battery_level_to_text(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: battery_level_to_text(data),
value_fn=lambda data: (data),
),
WeatherSensorEntityDescription(
key=WBGT_TEMP,

12
custom_components/sws12500/strings.json
View File

@ -73,16 +73,16 @@
"description": "Resend data to Pocasi Meteo CZ",
"title": "Configure Pocasi Meteo CZ",
"data": {
"POCASI_API_ID": "ID from your Pocasi Meteo APP",
"POCASI_API_KEY": "Key from your Pocasi Meteo APP",
"POCASI_SEND_INTERVAL": "Resend interval in seconds",
"POCASI_CZ_API_ID": "ID from your Pocasi Meteo APP",
"POCASI_CZ_API_KEY": "Key from your Pocasi Meteo APP",
"POCASI_CZ_SEND_INTERVAL": "Resend interval in seconds",
"pocasi_enabled_checkbox": "Enable resending data to Pocasi Meteo",
"pocasi_logger_checkbox": "Log data and responses"
},
"data_description": {
"POCASI_API_ID": "You can obtain your ID in Pocasi Meteo App",
"POCASI_API_KEY": "You can obtain your KEY in Pocasi Meteo App",
"POCASI_SEND_INTERVAL": "Resend interval in seconds (minimum 12s, default 30s)",
"POCASI_CZ_API_ID": "You can obtain your ID in Pocasi Meteo App",
"POCASI_CZ_API_KEY": "You can obtain your KEY in Pocasi Meteo App",
"POCASI_CZ_SEND_INTERVAL": "Resend interval in seconds (minimum 12s, default 30s)",
"pocasi_enabled_checkbox": "Enables resending data to Pocasi Meteo",
"pocasi_logger_checkbox": "Enable only if you want to send debbug data to the developer"
}

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

@ -73,16 +73,16 @@
"description": "Resend data to Pocasi Meteo CZ",
"title": "Configure Pocasi Meteo CZ",
"data": {
"POCASI_API_ID": "ID from your Pocasi Meteo APP",
"POCASI_API_KEY": "Key from your Pocasi Meteo APP",
"POCASI_SEND_INTERVAL": "Resend interval in seconds",
"POCASI_CZ_API_ID": "ID from your Pocasi Meteo APP",
"POCASI_CZ_API_KEY": "Key from your Pocasi Meteo APP",
"POCASI_CZ_SEND_INTERVAL": "Resend interval in seconds",
"pocasi_enabled_checkbox": "Enable resending data to Pocasi Meteo",
"pocasi_logger_checkbox": "Log data and responses"
},
"data_description": {
"POCASI_API_ID": "You can obtain your ID in Pocasi Meteo App",
"POCASI_API_KEY": "You can obtain your KEY in Pocasi Meteo App",
"POCASI_SEND_INTERVAL": "Resend interval in seconds (minimum 12s, default 30s)",
"POCASI_CZ_API_ID": "You can obtain your ID in Pocasi Meteo App",
"POCASI_CZ_API_KEY": "You can obtain your KEY in Pocasi Meteo App",
"POCASI_CZ_SEND_INTERVAL": "Resend interval in seconds (minimum 12s, default 30s)",
"pocasi_enabled_checkbox": "Enables resending data to Pocasi Meteo",
"pocasi_logger_checkbox": "Enable only if you want to send debbug data to the developer"
}

33
custom_components/sws12500/utils.py
View File

@ -10,17 +10,11 @@ import numpy as np
from homeassistant.components import persistent_notification
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
UnitOfPrecipitationDepth,
UnitOfTemperature,
UnitOfVolumetricFlux,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.translation import async_get_translations
from .const import (
AZIMUT,
BATTERY_LEVEL,
DATABASE_PATH,
DEV_DBG,
OUTSIDE_HUMIDITY,
@ -29,8 +23,8 @@ from .const import (
REMAP_WSLINK_ITEMS,
SENSORS_TO_LOAD,
WIND_SPEED,
UnitOfDir,
UnitOfBat,
UnitOfDir,
)
_LOGGER = logging.getLogger(__name__)
@ -55,7 +49,7 @@ async def translations(
)
if localize_key in _translations:
return _translations[localize_key]
return None
return ""
async def translated_notification(
@ -67,7 +61,7 @@ async def translated_notification(
*,
key: str = "message",
category: str = "notify",
) -> str:
):
"""Translate notification."""
localize_key = f"component.{translation_domain}.{category}.{translation_key}.{key}"
@ -98,7 +92,7 @@ async def translated_notification(
async def update_options(
hass: HomeAssistant, entry: ConfigEntry, update_key, update_value
) -> None:
) -> bool:
"""Update config.options entry."""
conf = {**entry.options}
conf[update_key] = update_value
@ -153,7 +147,7 @@ def check_disabled(
Returns list of found sensors or None
"""
log: bool = config_entry.options.get(DEV_DBG)
log: bool = config_entry.options.get(DEV_DBG, False)
entityFound: bool = False
_loaded_sensors = loaded_sensors(config_entry)
missing_sensors: list = []
@ -189,10 +183,15 @@ def battery_level_to_text(battery: int) -> UnitOfBat:
Returns UnitOfBat
"""
return {
level_map: dict[int, UnitOfBat] = {
0: UnitOfBat.LOW,
1: UnitOfBat.NORMAL,
}.get(int(battery) if battery is not None else None, UnitOfBat.UNKNOWN)
}
if battery is None:
return UnitOfBat.UNKNOWN
return level_map.get(int(battery), UnitOfBat.UNKNOWN)
def battery_level_to_icon(battery: UnitOfBat) -> str:
@ -219,7 +218,7 @@ def celsius_to_fahrenheit(celsius: float) -> float:
return celsius * 9.0 / 5.0 + 32
def heat_index(data: Any, convert: bool = False) -> UnitOfTemperature:
def heat_index(data: Any, convert: bool = False) -> float:
"""Calculate heat index from temperature.
data: dict with temperature and humidity
@ -257,7 +256,7 @@ def heat_index(data: Any, convert: bool = False) -> UnitOfTemperature:
return simple
def chill_index(data: Any, convert: bool = False) -> UnitOfTemperature:
def chill_index(data: Any, convert: bool = False) -> float:
"""Calculate wind chill index from temperature and wind speed.
data: dict with temperature and wind speed
@ -286,7 +285,7 @@ def chill_index(data: Any, convert: bool = False) -> UnitOfTemperature:
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
@ -314,7 +313,7 @@ def long_term_units_in_statistics_meta():
return sensor_units
async def migrate_data(hass: HomeAssistant, sensor_id: str | None = None) -> bool:
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)

8
custom_components/sws12500/windy_func.py
View File

@ -3,6 +3,8 @@
from datetime import datetime, timedelta
import logging
from aiohttp.client_exceptions import ClientError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -138,20 +140,20 @@ class WindyPush:
_LOGGER.critical(WINDY_INVALID_KEY)
text_for_test = WINDY_INVALID_KEY
update_options(self.hass, self.config, WINDY_ENABLED, False)
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 session.ClientError as ex:
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
update_options(self.hass, self.config, WINDY_ENABLED, False)
await update_options(self.hass, self.config, WINDY_ENABLED, False)
self.last_update = datetime.now()
self.next_update = self.last_update + timed(minutes=5)