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

First code commit. 

First release of code.
pull/1/head
schizza 2024-03-08 20:18:04 +01:00
parent 5d2bac671f
commit b0c7c72645
10 changed files with 1011 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
custom_components/sws12500/__init__.py Normal file
View File

@ -0,0 +1,143 @@
"""The Sencor SWS 12500 Weather Station integration."""
import logging
import aiohttp
from aiohttp.web_exceptions import HTTPUnauthorized
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import InvalidStateError, PlatformNotReady
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import API_ID, API_KEY, DEFAULT_URL, DOMAIN, WINDY_ENABLED
from .utils import remap_items
from .windy_func import WindyPush
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.SENSOR]
class IncorrectDataError(InvalidStateError):
"""Invalid exception."""
class WeatherDataUpdateCoordinator(DataUpdateCoordinator):
"""Manage fetched data."""
def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
"""Init global updater."""
self.hass = hass
self.config = config
self.windy = WindyPush(hass, config)
super().__init__(hass, _LOGGER, name=DOMAIN)
async def recieved_data(self, webdata):
"""Handle incoming data query."""
data = webdata.query
response = None
if "ID" not in data or "PASSWORD" not in data:
_LOGGER.error("Invalid request. No security data provided!")
raise HTTPUnauthorized
id_data = data["ID"]
key_data = data["PASSWORD"]
_id = self.config_entry.options.get(API_ID)
_key = self.config_entry.options.get(API_KEY)
if id_data != _id or key_data != _key:
_LOGGER.error("Unauthorised access!")
raise HTTPUnauthorized
if self.config_entry.options.get(WINDY_ENABLED):
response = await self.windy.push_data_to_windy(data)
self.async_set_updated_data(remap_items(data))
response = response if response else "OK"
return aiohttp.web.Response(body=f"{response}", status=200)
def register_path(
hass: HomeAssistant, url_path: str, coordinator: WeatherDataUpdateCoordinator
):
"""Register path to handle incoming data."""
try:
route = hass.http.app.router.add_route(
"GET", url_path, coordinator.recieved_data
)
except Exception: # pylint: disable=(broad-except)
_LOGGER.error("Unable to register URL handler!")
return False
_LOGGER.info(
"Registered path to handle weather data: %s",
route.get_info(), # pylint: disable=used-before-assignment
)
return True
def unregister_path():
"""Unregister path to handle incoming data."""
_LOGGER.error(
"Unable to delete webhook from API! Restart HA before adding integration!"
)
class Weather(WeatherDataUpdateCoordinator):
"""Weather class."""
def __init__(self, hass: HomeAssistant, config) -> None:
"""Init class."""
self.hass = hass
super().__init__(hass, config)
async def setup_update_listener(self, hass: HomeAssistant, entry: ConfigEntry):
"""Update setup listener."""
_LOGGER.info("Settings updated")
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the config entry for my device."""
coordinator = WeatherDataUpdateCoordinator(hass, entry)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = coordinator
weather = Weather(hass, entry)
if not register_path(hass, DEFAULT_URL, coordinator):
_LOGGER.error("Fatal: path not registered!")
raise PlatformNotReady
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(weather.setup_update_listener))
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if _ok:
hass.data[DOMAIN].pop(entry.entry_id)
unregister_path()
return _ok
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the component.
This component can only be configured through the Integrations UI.
"""
hass.data.setdefault(DOMAIN, {})
return True

163
custom_components/sws12500/config_flow.py Normal file
View File

@ -0,0 +1,163 @@
"""Config flow for Sencor SWS 12500 Weather Station integration."""
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from .const import (
API_ID,
API_KEY,
DOMAIN,
INVALID_CREDENTIALS,
WINDY_API_KEY,
WINDY_ENABLED,
WINDY_LOGGER_ENABLED,
)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(API_ID, default="API ID"): str,
vol.Required(API_KEY, default="API KEY"): str,
}
)
class CannotConnect(HomeAssistantError):
"""We can not connect. - not used in push mechanism."""
class InvalidAuth(HomeAssistantError):
"""Invalid auth exception."""
class ConfigOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle WeatherStation options."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
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"])
async def async_step_basic(self, user_input=None):
"""Manage basic options - credentials."""
errors = {}
api_id = self.config_entry.options.get(API_ID)
api_key = self.config_entry.options.get(API_KEY)
OPTIONAL_USER_DATA_SCHEMA = vol.Schema( # pylint: disable=invalid-name
{
vol.Required(API_ID, default=api_id): str,
vol.Required(API_KEY, default=api_key): str,
}
)
if user_input is None:
return self.async_show_form(
step_id="basic", data_schema=OPTIONAL_USER_DATA_SCHEMA, errors=errors
)
if user_input[API_ID] in INVALID_CREDENTIALS:
errors["base"] = "valid_credentials_api"
elif user_input[API_KEY] in INVALID_CREDENTIALS:
errors["base"] = "valid_credentials_key"
elif user_input[API_KEY] == user_input[API_ID]:
errors["base"] = "valid_credentials_match"
else:
# retain Windy options
data: dict = {}
data[WINDY_API_KEY] = self.config_entry.options.get(WINDY_API_KEY)
data[WINDY_ENABLED] = self.config_entry.options.get(WINDY_ENABLED)
data[WINDY_LOGGER_ENABLED] = self.config_entry.options.get(
WINDY_LOGGER_ENABLED
)
data.update(user_input)
return self.async_create_entry(title=DOMAIN, data=data)
# we are ending with error msg, reshow form
return self.async_show_form(
step_id="basic", data_schema=OPTIONAL_USER_DATA_SCHEMA, errors=errors
)
async def async_step_windy(self, user_input=None):
"""Manage windy options."""
errors = {}
windy_key = self.config_entry.options.get(WINDY_API_KEY)
windy_enabled = self.config_entry.options.get(WINDY_ENABLED)
windy_logger_enabled = self.config_entry.options.get(WINDY_LOGGER_ENABLED)
OPTIONAL_USER_DATA_SCHEMA = vol.Schema( # pylint: disable=invalid-name
{
vol.Optional(WINDY_API_KEY, default=windy_key): str,
vol.Optional(WINDY_ENABLED, default=windy_enabled): bool,
vol.Optional(WINDY_LOGGER_ENABLED, default=windy_logger_enabled): bool,
}
)
if user_input is None:
return self.async_show_form(
step_id="windy", data_schema=OPTIONAL_USER_DATA_SCHEMA, errors=errors
)
if (user_input[WINDY_ENABLED] is True) and (user_input[WINDY_API_KEY] == ""):
errors["base"] = "windy_key_required"
return self.async_show_form(
step_id="windy", data_schema=OPTIONAL_USER_DATA_SCHEMA, errors=errors
)
# retain API_ID and API_KEY from config
data: dict = {}
data[API_ID] = self.config_entry.options.get(API_ID)
data[API_KEY] = self.config_entry.options.get(API_KEY)
data.update(user_input)
return self.async_create_entry(title=DOMAIN, data=data)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Sencor SWS 12500 Weather Station."""
VERSION = 1
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
if user_input is None:
await self.async_set_unique_id(DOMAIN)
self._abort_if_unique_id_configured()
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
errors = {}
if user_input[API_ID] in INVALID_CREDENTIALS:
errors["base"] = "valid_credentials_api"
elif user_input[API_KEY] in INVALID_CREDENTIALS:
errors["base"] = "valid_credentials_key"
elif user_input[API_KEY] == user_input[API_ID]:
errors["base"] = "valid_credentials_match"
else:
return self.async_create_entry(
title=DOMAIN, data=user_input, options=user_input
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> ConfigOptionsFlowHandler:
"""Get the options flow for this handler."""
return ConfigOptionsFlowHandler(config_entry)

84
custom_components/sws12500/const.py Normal file
View File

@ -0,0 +1,84 @@
"""Constants."""
from typing import Final
DOMAIN = "sws12500"
DEFAULT_URL = "/weatherstation/updateweatherstation.php"
WINDY_URL = "https://stations.windy.com/pws/update/"
ICON = "mdi:weather"
API_KEY = "API_KEY"
API_ID = "API_ID"
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_SUCCESS: Final = (
"Windy successfully sent data and data was successfully inserted by Windy API"
)
WINDY_UNEXPECTED: Final = (
"Windy responded unexpectedly 3 times in a row. Send to Windy is now disabled!"
)
INVALID_CREDENTIALS: Final = [
"API",
"API_ID",
"API ID",
"_ID",
"ID",
"API KEY",
"API_KEY",
"KEY",
"_KEY",
]
PURGE_DATA: Final = [
"ID",
"PASSWORD",
"action",
"rtfreq",
"realtime",
"dateutc",
"solarradiation",
"indoortempf",
"indoorhumidity",
"dailyrainin",
]
BARO_PRESSURE: Final = "baro_pressure"
OUTSIDE_TEMP: Final = "outside_temp"
DEW_POINT: Final = "dew_point"
OUTSIDE_HUMIDITY: Final = "outside_humidity"
WIND_SPEED: Final = "wind_speed"
WIND_GUST: Final = "wind_gust"
WIND_DIR: Final = "wind_dir"
RAIN: Final = "rain"
DAILY_RAIN: Final = "daily_rain"
SOLAR_RADIATION: Final = "solar_radiation"
INDOOR_TEMP: Final = "indoor_temp"
INDOOR_HUMIDITY: Final = "indoor_humidity"
UV: Final = "uv"
CH2_TEMP: Final = "ch2_temp"
CH2_HUMIDITY: Final = "ch2_humidity"
REMAP_ITEMS: dict = {
"baromin": BARO_PRESSURE,
"tempf": OUTSIDE_TEMP,
"dewptf": DEW_POINT,
"humidity": OUTSIDE_HUMIDITY,
"windspeedmph": WIND_SPEED,
"windgustmph": WIND_GUST,
"winddir": WIND_DIR,
"rainin": RAIN,
"dailyrainin": DAILY_RAIN,
"solarradiation": SOLAR_RADIATION,
"indoortempf": INDOOR_TEMP,
"indoorhumidity": INDOOR_HUMIDITY,
"UV": UV,
"soiltempf": CH2_TEMP,
"soilmoisture": CH2_HUMIDITY,
}

14
custom_components/sws12500/manifest.json Normal file
View File

@ -0,0 +1,14 @@
{
"domain": "sws12500",
"name": "Sencor SWS 12500 Weather Station",
"codeowners": ["@schizza"],
"config_flow": true,
"dependencies": ["http"],
"documentation": "https://www.home-assistant.io/integrations/sws12500",
"homekit": {},
"iot_class": "local_push",
"requirements": [],
"ssdp": [],
"version": "0.0.2",
"zeroconf": []
}

46
custom_components/sws12500/property.py Normal file
View File

@ -0,0 +1,46 @@
@property
def translation_key(self):
"""Return translation key."""
return self.entity_description.translation_key
@property
def device_class(self):
"""Return device class."""
return self.entity_description.device_class
@property
def name(self) -> str:
"""Return the name of the switch."""
return str(self.entity_description.name)
@property
def unique_id(self) -> str:
"""Return a unique, Home Assistant friendly identifier for this entity."""
return self.entity_description.key
@property
def native_value(self):
"""Return value of entity."""
return self._state
@property
def icon(self) -> str:
"""Return icon of entity."""
return str(self.entity_description.icon)
@property
def native_unit_of_measurement(self) -> str:
"""Return unit of measurement."""
return str(self.entity_description.native_unit_of_measurement)
@property
def state_class(self) -> str:
"""Return stateClass."""
return str(self.entity_description.state_class)
@property
def suggested_unit_of_measurement(self) -> str:
"""Return sugestet_unit_of_measurement."""
return str(self.entity_description.suggested_unit_of_measurement)

262
custom_components/sws12500/sensor.py Normal file
View File

@ -0,0 +1,262 @@
"""Sensors definition for SWS12500."""
from dataclasses import dataclass
from typing import Any
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntityDescription,
SensorEntity,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
UnitOfIrradiance,
UnitOfPrecipitationDepth,
UnitOfPressure,
UnitOfSpeed,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import WeatherDataUpdateCoordinator
from .const import (
BARO_PRESSURE,
DAILY_RAIN,
DEW_POINT,
DOMAIN,
INDOOR_HUMIDITY,
INDOOR_TEMP,
OUTSIDE_HUMIDITY,
OUTSIDE_TEMP,
RAIN,
SOLAR_RADIATION,
UV,
WIND_DIR,
WIND_GUST,
WIND_SPEED,
)
@dataclass
class WeatherSensorEntityDescription(SensorEntityDescription):
"""Describe Weather Sensor entities."""
SENSOR_TYPES: tuple[WeatherSensorEntityDescription, ...] = (
WeatherSensorEntityDescription(
key=INDOOR_TEMP,
name="Indoor temperature",
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:thermometer",
device_class=SensorDeviceClass.TEMPERATURE,
),
WeatherSensorEntityDescription(
key=INDOOR_HUMIDITY,
name="Indoor humidity",
native_unit_of_measurement="%",
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:thermometer",
device_class=SensorDeviceClass.HUMIDITY,
),
WeatherSensorEntityDescription(
key=OUTSIDE_TEMP,
name="Outside Temperature",
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:thermometer",
device_class=SensorDeviceClass.TEMPERATURE,
),
WeatherSensorEntityDescription(
key=OUTSIDE_HUMIDITY,
name="Outside humidity",
native_unit_of_measurement="%",
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:thermometer",
device_class=SensorDeviceClass.HUMIDITY,
),
WeatherSensorEntityDescription(
key=DEW_POINT,
name="Dew point",
native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:thermometer-lines",
device_class=SensorDeviceClass.TEMPERATURE,
),
WeatherSensorEntityDescription(
key=BARO_PRESSURE,
name="Barometric pressure",
native_unit_of_measurement=UnitOfPressure.INHG,
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:thermometer-lines",
device_class=SensorDeviceClass.ATMOSPHERIC_PRESSURE,
suggested_unit_of_measurement=UnitOfPressure.HPA,
),
WeatherSensorEntityDescription(
key=WIND_SPEED,
name="Wind speed",
native_unit_of_measurement=UnitOfSpeed.MILES_PER_HOUR,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.WIND_SPEED,
suggested_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
icon="mdi:weather-windy",
),
WeatherSensorEntityDescription(
key=WIND_GUST,
name="Wind gust",
native_unit_of_measurement=UnitOfSpeed.MILES_PER_HOUR,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.WIND_SPEED,
suggested_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
icon="mdi:windsock",
),
WeatherSensorEntityDescription(
key=WIND_DIR,
name="Wind direction",
native_unit_of_measurement="°",
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:sign-direction",
),
WeatherSensorEntityDescription(
key=RAIN,
name="Rain",
native_unit_of_measurement=UnitOfPrecipitationDepth.INCHES,
device_class=SensorDeviceClass.PRECIPITATION,
state_class=SensorStateClass.TOTAL,
suggested_unit_of_measurement=UnitOfPrecipitationDepth.MILLIMETERS,
suggested_display_precision=2,
icon="mdi:weather-pouring",
),
WeatherSensorEntityDescription(
key=DAILY_RAIN,
name="Daily precipitation",
native_unit_of_measurement="in/d",
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.PRECIPITATION_INTENSITY,
suggested_unit_of_measurement="mm/h",
suggested_display_precision=2,
icon="mdi:weather-pouring",
),
WeatherSensorEntityDescription(
key=SOLAR_RADIATION,
name="Solar irradiance",
native_unit_of_measurement=UnitOfIrradiance.WATTS_PER_SQUARE_METER,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.IRRADIANCE,
icon="mdi:weather-sunny",
),
WeatherSensorEntityDescription(
key=UV,
name="UV index",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="",
icon="mdi:sunglasses",
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info=None,
) -> None:
"""Set up Weather Station sensors."""
coordinator: WeatherDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
sensors = []
for description in SENSOR_TYPES:
sensors.append(WeatherSensor(hass, description, coordinator))
async_add_entities(sensors)
class WeatherSensor(CoordinatorEntity[WeatherDataUpdateCoordinator], SensorEntity):
"""Implementation of Weather Sensor entity."""
entity_description: WeatherSensorEntityDescription
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
hass: HomeAssistant,
description: WeatherSensorEntityDescription,
coordinator: WeatherDataUpdateCoordinator,
) -> None:
"""Initialize sensor."""
super().__init__(coordinator)
self.hass = hass
self.coordinator = coordinator
self.entity_description = description
self._state: StateType = None
self._attrs: dict[str, Any] = {}
self._available = False
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._state = self.coordinator.data.get(self.entity_description.key)
self.async_write_ha_state()
self.async_registry_entry_updated()
@property
def device_class(self) -> SensorDeviceClass:
"""Return device class."""
return self.entity_description.device_class
@property
def name(self) -> str:
"""Return the name of the switch."""
return str(self.entity_description.name)
@property
def unique_id(self) -> str:
"""Return a unique, Home Assistant friendly identifier for this entity."""
return self.entity_description.key
@property
def native_value(self) -> None:
"""Return value of entity."""
return self._state
@property
def icon(self) -> str:
"""Return icon of entity."""
return str(self.entity_description.icon)
@property
def native_unit_of_measurement(self) -> str:
"""Return unit of measurement."""
return str(self.entity_description.native_unit_of_measurement)
@property
def state_class(self) -> str:
"""Return stateClass."""
return str(self.entity_description.state_class)
@property
def suggested_unit_of_measurement(self) -> str:
"""Return sugestet_unit_of_measurement."""
return str(self.entity_description.suggested_unit_of_measurement)
@property
def device_info(self) -> DeviceInfo:
"""Device info."""
return DeviceInfo(
connections=set(),
name="Weather Station SWS 12500",
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN,)}, # type: ignore[arg-type]
manufacturer="Schizza",
model="Weather Station SWS 12500",
)

54
custom_components/sws12500/strings.json Normal file
View File

@ -0,0 +1,54 @@
{
"config": {
"step": {
"user": {
"title": "Configure access for Weather Station",
"description": "Provide API ID and API KEY so the Weather Station can access HomeAssistant",
"data": {
"api_id": "API ID / Station ID",
"api_key": "API KEY / Password"
}
}
},
"error": {
"valid_credentials_api": "Provide valid API ID.",
"valid_credentials_key": "Provide valid API KEY.",
"valid_credentials_match": "API ID and API KEY should not be the same."
}
},
"options": {
"step": {
"init": {
"title": "Configure SWS12500 Integration",
"description": "Choose what do you want to configure. If basic access or resending data for Windy site",
"menu_options": {
"basic": "Basic - configure credentials for Weather Station",
"windy": "Windy configuration"
}
},
"basic": {
"title": "Configure credentials",
"description": "Provide API ID and API KEY so the Weather Station can access HomeAssistant",
"data": {
"API_ID": "API ID / Station ID",
"API_KEY": "API KEY / Password"
}
},
"windy": {
"title": "Configure Windy",
"description": "Resend weather data to your Windy stations.",
"data": {
"WINDY_API_KEY": "API KEY provided by Windy",
"windy_enabled_checkbox": "Enable resending data to Windy",
"windy_logger_checkbox": "Log Windy data and responses"
}
}
},
"error": {
"valid_credentials_api": "Provide valid API ID.",
"valid_credentials_key": "Provide valid API KEY.",
"valid_credentials_match": "API ID and API KEY should not be the same.",
"windy_key_required": "Windy API key is required if you want to enable this function."
}
}
}

54
custom_components/sws12500/translations/en.json Normal file
View File

@ -0,0 +1,54 @@
{
"config": {
"error": {
"valid_credentials_api": "Provide valid API ID.",
"valid_credentials_key": "Provide valid API KEY.",
"valid_credentials_match": "API ID and API KEY should not be the same."
},
"step": {
"user": {
"data": {
"api_id": "API ID / Station ID",
"api_key": "API KEY / Password"
},
"description": "Provide API ID and API KEY so the Weather Station can access HomeAssistant",
"title": "Configure access for Weather Station"
}
}
},
"options": {
"error": {
"valid_credentials_api": "Provide valid API ID.",
"valid_credentials_key": "Provide valid API KEY.",
"valid_credentials_match": "API ID and API KEY should not be the same.",
"windy_key_required": "Windy API key is required if you want to enable this function."
},
"step": {
"basic": {
"data": {
"API_ID": "API ID / Station ID",
"API_KEY": "API KEY / Password"
},
"description": "Provide API ID and API KEY so the Weather Station can access HomeAssistant",
"title": "Configure credentials"
},
"init": {
"description": "Choose what do you want to configure. If basic access or resending data for Windy site",
"menu_options": {
"basic": "Basic - configure credentials for Weather Station",
"windy": "Windy configuration"
},
"title": "Configure SWS12500 Integration"
},
"windy": {
"data": {
"WINDY_API_KEY": "API KEY provided by Windy",
"windy_enabled_checkbox": "Enable resending data to Windy",
"windy_logger_checkbox": "Log Windy data and responses"
},
"description": "Resend weather data to your Windy stations.",
"title": "Configure Windy"
}
}
}
}

30
custom_components/sws12500/utils.py Normal file
View File

@ -0,0 +1,30 @@
"""Utils for SWS12500."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import REMAP_ITEMS
def update_options(
hass: HomeAssistant, entry: ConfigEntry, update_key, update_value
) -> None:
"""Update config.options entry."""
conf = {}
for k in entry.options:
conf[k] = entry.options[k]
conf[update_key] = update_value
hass.config_entries.async_update_entry(entry, options=conf)
def remap_items(entities):
"""Remap items in query."""
items = {}
for item in entities:
if item in REMAP_ITEMS:
items[REMAP_ITEMS[item]] = entities[item]
return items

161
custom_components/sws12500/windy_func.py Normal file
View File

@ -0,0 +1,161 @@
"""Windy functions."""
from datetime import datetime, timedelta
import logging
import aiohttp
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import (
PURGE_DATA,
WINDY_API_KEY,
WINDY_ENABLED,
WINDY_INVALID_KEY,
WINDY_LOGGER_ENABLED,
WINDY_NOT_INSERTED,
WINDY_SUCCESS,
WINDY_UNEXPECTED,
WINDY_URL,
)
from .utils import update_options
_LOGGER = logging.getLogger(__name__)
RESPONSE_FOR_TEST = False
class WindyNotInserted(Exception):
"""NotInserted state."""
class WindySuccess(Exception):
"""WindySucces state."""
class WindyApiKeyError(Exception):
"""Windy API Key error."""
def timed(minutes: int):
"""Simulate timedelta.
So we can mock td in tests.
"""
return timedelta(minutes=minutes)
class WindyPush:
"""Push data to Windy."""
def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
"""Init."""
self.hass = hass
self.config = config
""" lets wait for 1 minute to get initial data from station
and then try to push first data to Windy
"""
self.last_update = datetime.now()
self.next_update = datetime.now() + timed(minutes=1)
self.log = self.config.options.get(WINDY_LOGGER_ENABLED)
self.invalid_response_count = 0
def verify_windy_response( # pylint: disable=useless-return
self,
response: str,
) -> WindyNotInserted | WindySuccess | WindyApiKeyError | None:
"""Verify answer form Windy."""
if "NOTICE" in response:
raise WindyNotInserted
if "SUCCESS" in response:
raise WindySuccess
if "Invalid API key" in response:
raise WindyApiKeyError
return None
async def push_data_to_windy(self, data):
"""Pushes weather data do Windy stations.
Interval is 5 minutes, otherwise Windy would not accepts data.
we are sending almost the same data as we received
from station. But we need to do some clean up.
"""
text_for_test = None
if self.log:
_LOGGER.info(
"Windy last update = %s, next update at: %s",
str(self.last_update),
str(self.next_update),
)
if self.next_update > datetime.now():
return False
purged_data = dict(data)
for purge in PURGE_DATA:
if purge in purged_data:
purged_data.pop(purge)
if "dewptf" in purged_data:
dewpoint = round(((float(purged_data.pop("dewptf")) - 32) / 1.8), 1)
purged_data["dewpoint"] = str(dewpoint)
windy_api_key = self.config.options.get(WINDY_API_KEY)
request_url = f"{WINDY_URL}{windy_api_key}"
if self.log:
_LOGGER.info("Dataset for windy: %s", purged_data)
async with aiohttp.ClientSession(
connector=aiohttp.TCPConnector(ssl=False), trust_env=True
) as session: # verify_ssl=False; intended to be False
try:
async with session.get(request_url, params=purged_data) as resp:
status = await resp.text()
try:
self.verify_windy_response(status)
except WindyNotInserted:
# log despite of settings
_LOGGER.error(WINDY_NOT_INSERTED)
text_for_test = WINDY_NOT_INSERTED
except WindyApiKeyError:
# log despite of settings
_LOGGER.critical(WINDY_INVALID_KEY)
text_for_test = WINDY_INVALID_KEY
update_options(self.hass, self.config, WINDY_ENABLED, False)
except WindySuccess:
if self.log:
_LOGGER.info(WINDY_SUCCESS)
text_for_test = WINDY_SUCCESS
except aiohttp.ClientConnectionError 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)
self.last_update = datetime.now()
self.next_update = self.last_update + timed(minutes=5)
if self.log:
_LOGGER.info("Next update: %s", str(self.next_update))
if RESPONSE_FOR_TEST and text_for_test:
return text_for_test
return None