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 a1f2bf10ea into b1cec2f38f

FerronN 2025-08-22 16:15:14 +00:00 committed by GitHub
parent b1cec2f38f a1f2bf10ea
commit a9534a3abd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 131 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
custom_components/sws12500/config_flow.py
View File

@ -7,7 +7,6 @@ import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, OptionsFlow
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from .utils import long_term_units_in_statistics_meta, migrate_data
from .const import (
API_ID,
@ -16,7 +15,6 @@ from .const import (
DOMAIN,
INVALID_CREDENTIALS,
SENSORS_TO_LOAD,
SENSOR_TO_MIGRATE,
WINDY_API_KEY,
WINDY_ENABLED,
WINDY_LOGGER_ENABLED,
@ -41,7 +39,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
self.windy_data: dict[str, Any] = {}
self.windy_data_schema = {}
self.user_data: dict[str, str] = {}
self.user_data: dict[str, Any] = {}
self.user_data_schema = {}
self.sensors: dict[str, Any] = {}
self.migrate_schema = {}
@ -50,7 +48,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
def config_entry(self):
return self.hass.config_entries.async_get_entry(self.handler)
def _get_entry_data(self):
async def _get_entry_data(self):
"""Get entry data."""
self.user_data: dict[str, Any] = {
@ -61,16 +59,18 @@ class ConfigOptionsFlowHandler(OptionsFlow):
}
self.user_data_schema = {
vol.Required(API_ID, default=self.user_data[API_ID] or ""): str,
vol.Required(API_KEY, default=self.user_data[API_KEY] or ""): str,
vol.Optional(WSLINK, default=self.user_data[WSLINK]): bool or False,
vol.Optional(DEV_DBG, default=self.user_data[DEV_DBG]): bool or False,
vol.Required(API_ID, default=self.user_data.get(API_ID, "")): str,
vol.Required(API_KEY, default=self.user_data.get(API_KEY, "")): str,
vol.Optional(WSLINK, default=self.user_data.get(WSLINK, False)): bool,
vol.Optional(DEV_DBG, default=self.user_data.get(DEV_DBG, False)): bool,
}
self.sensors: dict[str, Any] = {
SENSORS_TO_LOAD: self.config_entry.options.get(SENSORS_TO_LOAD)
if isinstance(self.config_entry.options.get(SENSORS_TO_LOAD), list)
else []
SENSORS_TO_LOAD: (
self.config_entry.options.get(SENSORS_TO_LOAD)
if isinstance(self.config_entry.options.get(SENSORS_TO_LOAD), list)
else []
)
}
self.windy_data: dict[str, Any] = {
@ -83,34 +83,26 @@ class ConfigOptionsFlowHandler(OptionsFlow):
self.windy_data_schema = {
vol.Optional(
WINDY_API_KEY, default=self.windy_data[WINDY_API_KEY] or ""
WINDY_API_KEY, default=self.windy_data.get(WINDY_API_KEY, "")
): str,
vol.Optional(WINDY_ENABLED, default=self.windy_data[WINDY_ENABLED]): bool
or False,
vol.Optional(
WINDY_LOGGER_ENABLED,
default=self.windy_data[WINDY_LOGGER_ENABLED],
): bool or False,
}
self.migrate_schema = {
vol.Required(SENSOR_TO_MIGRATE): vol.In(
long_term_units_in_statistics_meta() or {}
),
vol.Optional("trigger_action", default=False): bool,
): bool
or False,
}
async def async_step_init(self, user_input=None):
"""Manage the options - show menu first."""
return self.async_show_menu(
step_id="init", menu_options=["basic", "windy", "migration"]
)
return self.async_show_menu(step_id="init", menu_options=["basic", "windy"])
async def async_step_basic(self, user_input=None):
"""Manage basic options - credentials."""
errors = {}
self._get_entry_data()
await self._get_entry_data()
if user_input is None:
return self.async_show_form(
@ -147,7 +139,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
"""Manage windy options."""
errors = {}
self._get_entry_data()
await self._get_entry_data()
if user_input is None:
return self.async_show_form(
@ -160,7 +152,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
errors[WINDY_API_KEY] = "windy_key_required"
return self.async_show_form(
step_id="windy",
data_schema=self.windy_data_schema,
data_schema=vol.Schema(self.windy_data_schema),
errors=errors,
)
@ -172,51 +164,6 @@ class ConfigOptionsFlowHandler(OptionsFlow):
return self.async_create_entry(title=DOMAIN, data=user_input)
async def async_step_migration(self, user_input=None):
"""Migrate sensors."""
# hj
errors = {}
self._get_entry_data()
if user_input is None:
return self.async_show_form(
step_id="migration",
data_schema=vol.Schema(self.migrate_schema),
errors=errors,
description_placeholders={
"migration_status": "-",
"migration_count": "-",
},
)
if user_input.get("trigger_action"):
# Akce se vykoná po zaškrtnutí
count = await self.hass.async_add_executor_job(
migrate_data, user_input.get(SENSOR_TO_MIGRATE)
)
return self.async_show_form(
step_id="migration",
data_schema=vol.Schema(self.migrate_schema),
errors=errors,
description_placeholders={
"migration_status": user_input.get(SENSOR_TO_MIGRATE),
"migration_count": count,
},
)
# retain windy data
user_input.update(self.windy_data)
# retain user_data
user_input.update(self.user_data)
# retain senors
user_input.update(self.sensors)
return self.async_create_entry(title=DOMAIN, data=user_input)
class ConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Sencor SWS 12500 Weather Station."""

26
custom_components/sws12500/const.py
View File

@ -23,8 +23,12 @@ WSLINK: Final = "wslink"
WINDY_API_KEY = "WINDY_API_KEY"
WINDY_ENABLED: Final = "windy_enabled_checkbox"
WINDY_LOGGER_ENABLED: Final = "windy_logger_checkbox"
WINDY_NOT_INSERTED: Final = "Data was succefuly sent to Windy, but not inserted by Windy API. Does anyone else sent data to Windy?"
WINDY_INVALID_KEY: Final = "Windy API KEY is invalid. Send data to Windy is now disabled. Check your API KEY and try again."
WINDY_NOT_INSERTED: Final = (
"Data was succefuly sent to Windy, but not inserted by Windy API. Does anyone else sent data to Windy?"
)
WINDY_INVALID_KEY: Final = (
"Windy API KEY is invalid. Send data to Windy is now disabled. Check your API KEY and try again."
)
WINDY_SUCCESS: Final = (
"Windy successfully sent data and data was successfully inserted by Windy API"
)
@ -63,6 +67,7 @@ OUTSIDE_TEMP: Final = "outside_temp"
DEW_POINT: Final = "dew_point"
OUTSIDE_HUMIDITY: Final = "outside_humidity"
OUTSIDE_CONNECTION: Final = "outside_connection"
OUTSIDE_BATTERY: Final = "outside_battery"
WIND_SPEED: Final = "wind_speed"
WIND_GUST: Final = "wind_gust"
WIND_DIR: Final = "wind_dir"
@ -139,6 +144,7 @@ REMAP_WSLINK_ITEMS: dict = {
"t1rainyr": YEARLY_RAIN,
"t234c2tem": CH3_TEMP,
"t234c2hum": CH3_HUMIDITY,
"t1bat": OUTSIDE_BATTERY,
}
# TODO: Add more sensors
@ -155,6 +161,7 @@ DISABLED_BY_DEFAULT: Final = [
CH3_HUMIDITY,
CH4_TEMP,
CH4_HUMIDITY,
OUTSIDE_BATTERY,
]
@ -197,3 +204,18 @@ AZIMUT: list[UnitOfDir] = [
UnitOfDir.NNW,
UnitOfDir.N,
]
class UnitOfBat(StrEnum):
"""Battery level unit of measure."""
LOW = "low"
NORMAL = "normal"
UNKNOWN = "unknown"
BATTERY_LEVEL: list[UnitOfBat] = [
UnitOfBat.LOW,
UnitOfBat.NORMAL,
UnitOfBat.UNKNOWN,
]

19
custom_components/sws12500/sensor.py
View File

@ -15,6 +15,7 @@ from .const import (
CHILL_INDEX,
DOMAIN,
HEAT_INDEX,
OUTSIDE_BATTERY,
OUTSIDE_HUMIDITY,
OUTSIDE_TEMP,
SENSORS_TO_LOAD,
@ -26,7 +27,7 @@ from .const import (
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
from .utils import chill_index, heat_index, battery_level_to_icon, battery_level_to_text
_LOGGER = logging.getLogger(__name__)
@ -130,13 +131,27 @@ class WeatherSensor(
):
return self.entity_description.value_fn(chill_index(self.coordinator.data))
return None if self._data == "" else self.entity_description.value_fn(self._data)
return (
None if self._data == "" else self.entity_description.value_fn(self._data)
)
@property
def suggested_entity_id(self) -> str:
"""Return name."""
return generate_entity_id("sensor.{}", self.entity_description.key)
@property
def icon(self) -> str | None:
"""Return the dynamic icon for battery representation."""
if self.entity_description.key == OUTSIDE_BATTERY:
try:
return battery_level_to_icon(self.native_value)
except Exception:
return "mdi:battery-unknown"
return self.entity_description.icon
@property
def device_info(self) -> DeviceInfo:
"""Device info."""

20
custom_components/sws12500/sensors_wslink.py
View File

@ -27,25 +27,26 @@ from .const import (
DAILY_RAIN,
DEW_POINT,
HEAT_INDEX,
HOURLY_RAIN,
INDOOR_HUMIDITY,
INDOOR_TEMP,
MONTHLY_RAIN,
OUTSIDE_BATTERY,
OUTSIDE_HUMIDITY,
OUTSIDE_TEMP,
RAIN,
SOLAR_RADIATION,
UV,
WEEKLY_RAIN,
WIND_AZIMUT,
WIND_DIR,
WIND_GUST,
WIND_SPEED,
UnitOfDir,
MONTHLY_RAIN,
YEARLY_RAIN,
HOURLY_RAIN,
WEEKLY_RAIN,
UnitOfDir,
)
from .sensors_common import WeatherSensorEntityDescription
from .utils import wind_dir_to_text
from .utils import battery_level_to_icon, battery_level_to_text, wind_dir_to_text
SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
WeatherSensorEntityDescription(
@ -143,7 +144,7 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
WeatherSensorEntityDescription(
key=RAIN,
native_unit_of_measurement=UnitOfVolumetricFlux.MILLIMETERS_PER_HOUR,
device_class=SensorDeviceClass.PRECIPITATION,
device_class=SensorDeviceClass.PRECIPITATION_INTENSITY,
state_class=SensorStateClass.TOTAL,
suggested_unit_of_measurement=UnitOfVolumetricFlux.MILLIMETERS_PER_HOUR,
suggested_display_precision=2,
@ -303,4 +304,11 @@ SENSOR_TYPES_WSLINK: tuple[WeatherSensorEntityDescription, ...] = (
translation_key=CHILL_INDEX,
value_fn=lambda data: cast("int", data),
),
WeatherSensorEntityDescription(
key=OUTSIDE_BATTERY,
translation_key=OUTSIDE_BATTERY,
icon="mdi:battery-unknown",
device_class=SensorDeviceClass.ENUM,
value_fn=lambda data: battery_level_to_text(data),
),
)

8
custom_components/sws12500/strings.json
View File

@ -131,6 +131,14 @@
"wnw": "WNW",
"nw": "NW",
"nnw": "NNW"
},
"outside_battery": {
"name": "Outside battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
}
}
}

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

@ -136,6 +136,14 @@
"nw": "SZ",
"nnw": "SSZ"
}
},
"outside_battery": {
"name": "Stav nabití venkovní baterie",
"state": {
"low": "Nízká",
"normal": "Normální",
"unknown": "Neznámá / zcela vybitá"
}
}
}
},

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

@ -136,6 +136,14 @@
"nw": "NW",
"nnw": "NNW"
}
},
"outside_battery": {
"name": "Outside battery level",
"state": {
"normal": "OK",
"low": "Low",
"unknown": "Unknown / drained out"
}
}
}
},

34
custom_components/sws12500/utils.py
View File

@ -20,6 +20,7 @@ from homeassistant.helpers.translation import async_get_translations
from .const import (
AZIMUT,
BATTERY_LEVEL,
DATABASE_PATH,
DEV_DBG,
OUTSIDE_HUMIDITY,
@ -29,6 +30,7 @@ from .const import (
SENSORS_TO_LOAD,
WIND_SPEED,
UnitOfDir,
UnitOfBat,
)
_LOGGER = logging.getLogger(__name__)
@ -181,6 +183,32 @@ def wind_dir_to_text(deg: float) -> UnitOfDir | None:
return None
def battery_level_to_text(battery: int) -> UnitOfBat:
"""Return battery level in text representation.
Returns UnitOfBat
"""
return {
0: UnitOfBat.LOW,
1: UnitOfBat.NORMAL,
}.get(int(battery) if battery is not None else None, 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
@ -267,10 +295,12 @@ def long_term_units_in_statistics_meta():
db = conn.cursor()
try:
db.execute("""
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