Merge a3dc3d0d53 into 9f36ab5d4c

- Add typing and runtime checks with py_typecheck for config options and
  incoming data
- Enhance authentication validation and error handling in data
  coordinator
- Refactor Routes class with better dispatch logic and route
  enabling/disabling
- Clean up async functions and improve logging consistency
- Add type hints and annotations throughout the integration
- Update manifest to include typecheck-runtime dependency

custom_components/sws12500/__init__.py

@@ -1,14 +1,20 @@
 """The Sencor SWS 12500 Weather Station integration."""
 
 import logging
+from typing import Any
 
 import aiohttp.web
 from aiohttp.web_exceptions import HTTPUnauthorized
+from py_typecheck import checked, checked_or
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import InvalidStateError, PlatformNotReady
+from homeassistant.exceptions import (
+    ConfigEntryNotReady,
+    InvalidStateError,
+    PlatformNotReady,
+)
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from .const import (
@@ -24,7 +30,7 @@ from .const import (
     WSLINK_URL,
 )
 from .pocasti_cz import PocasiPush
-from .routes import Routes, unregistred
+from .routes import Routes
 from .utils import (
     anonymize,
     check_disabled,
@@ -50,19 +56,20 @@ class WeatherDataUpdateCoordinator(DataUpdateCoordinator):
 
     def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
         """Init global updater."""
-        self.hass = hass
-        self.config = config
-        self.windy = WindyPush(hass, config)
+        self.hass: HomeAssistant = hass
+        self.config: ConfigEntry = config
+        self.windy: WindyPush = WindyPush(hass, config)
         self.pocasi: PocasiPush = PocasiPush(hass, config)
         super().__init__(hass, _LOGGER, name=DOMAIN)
 
-    async def recieved_data(self, webdata):
+    async def recieved_data(self, webdata: aiohttp.web.Request) -> aiohttp.web.Response:
         """Handle incoming data query."""
-        _wslink = self.config_entry.options.get(WSLINK)
-        data = webdata.query
 
-        response = None
+        _wslink: bool = checked_or(self.config.options.get(WSLINK), bool, False)
 
+        data: dict[str, Any] = dict(webdata.query)
+
+        # Check if station is sending auth data
         if not _wslink and ("ID" not in data or "PASSWORD" not in data):
             _LOGGER.error("Invalid request. No security data provided!")
             raise HTTPUnauthorized
@@ -71,44 +78,67 @@ class WeatherDataUpdateCoordinator(DataUpdateCoordinator):
             _LOGGER.error("Invalid request. No security data provided!")
             raise HTTPUnauthorized
 
-        if _wslink:
-            id_data = data["wsid"]
-            key_data = data["wspw"]
-        else:
-            id_data = data["ID"]
-            key_data = data["PASSWORD"]
+        id_data: str = ""
+        key_data: str = ""
 
-        _id = self.config_entry.options.get(API_ID)
-        _key = self.config_entry.options.get(API_KEY)
+        if _wslink:
+            id_data = data.get("wsid", "")
+            key_data = data.get("wspw", "")
+        else:
+            id_data = data.get("ID", "")
+            key_data = data.get("PASSWORD", "")
+
+        # Check if we have valid auth data in the integration
+
+        if (_id := checked(self.config.options.get(API_ID), str)) is None:
+            _LOGGER.error("We don't have API ID set! Update your config!")
+            raise IncorrectDataError
+
+        if (_key := checked(self.config.options.get(API_KEY), str)) is None:
+            _LOGGER.error("We don't have API KEY set! Update your config!")
+            raise IncorrectDataError
 
         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)
+        if self.config.options.get(WINDY_ENABLED, False):
+            await self.windy.push_data_to_windy(data)
 
-        if self.config.options.get(POCASI_CZ_ENABLED):
+        if self.config.options.get(POCASI_CZ_ENABLED, False):
             await self.pocasi.push_data_to_server(data, "WSLINK" if _wslink else "WU")
 
-        remaped_items = (
-            remap_wslink_items(data)
-            if self.config_entry.options.get(WSLINK)
-            else remap_items(data)
+        remaped_items: dict[str, str] = (
+            remap_wslink_items(data) if _wslink else remap_items(data)
         )
 
-        if sensors := check_disabled(self.hass, remaped_items, self.config):
-            translate_sensors = [
+        if sensors := check_disabled(remaped_items, self.config):
+            if (
+                translate_sensors := checked(
+                    [
                         await translations(
-                    self.hass, DOMAIN, f"sensor.{t_key}", key="name", category="entity"
+                            self.hass,
+                            DOMAIN,
+                            f"sensor.{t_key}",
+                            key="name",
+                            category="entity",
                         )
                         for t_key in sensors
                         if await translations(
-                    self.hass, DOMAIN, f"sensor.{t_key}", key="name", category="entity"
+                            self.hass,
+                            DOMAIN,
+                            f"sensor.{t_key}",
+                            key="name",
+                            category="entity",
                         )
                         is not None
-            ]
-            human_readable = "\n".join(translate_sensors)
+                    ],
+                    list[str],
+                )
+            ) is not None:
+                human_readable: str = "\n".join(translate_sensors)
+            else:
+                human_readable = ""
 
             await translated_notification(
                 self.hass,
@@ -126,82 +156,42 @@ class WeatherDataUpdateCoordinator(DataUpdateCoordinator):
         if self.config_entry.options.get(DEV_DBG):
             _LOGGER.info("Dev log: %s", anonymize(data))
 
-        response = response or "OK"
-        return aiohttp.web.Response(body=f"{response or 'OK'}", status=200)
+        return aiohttp.web.Response(body="OK", status=200)
 
 
 def register_path(
     hass: HomeAssistant,
-    url_path: str,
     coordinator: WeatherDataUpdateCoordinator,
     config: ConfigEntry,
-):
-    """Register path to handle incoming data."""
+) -> bool:
+    """Register paths to webhook."""
 
-    hass_data = hass.data.setdefault(DOMAIN, {})
-    debug = config.options.get(DEV_DBG)
-    _wslink = config.options.get(WSLINK, False)
+    hass.data.setdefault(DOMAIN, {})
+    if (hass_data := checked(hass.data[DOMAIN], dict[str, Any])) is None:
+        raise ConfigEntryNotReady
 
-    routes: Routes = hass_data.get("routes", Routes())
+    _wslink: bool = checked_or(config.options.get(WSLINK), bool, False)
 
-    if not routes.routes:
-        routes = Routes()
-        _LOGGER.info("Routes not found, creating new routes")
-
-        if debug:
-            _LOGGER.debug("Enabled route is: %s, WSLink is %s", url_path, _wslink)
+    # Create internal route dispatcher with provided urls
+    routes: Routes = Routes()
+    routes.add_route(DEFAULT_URL, coordinator.recieved_data, enabled=not _wslink)
+    routes.add_route(WSLINK_URL, coordinator.recieved_data, enabled=_wslink)
 
+    # Register webhooks in HomeAssistant with dispatcher
     try:
-            default_route = hass.http.app.router.add_get(
-                DEFAULT_URL,
-                coordinator.recieved_data if not _wslink else unregistred,
-                name="weather_default_url",
-            )
-            if debug:
-                _LOGGER.debug("Default route: %s", default_route)
-
-            wslink_route = hass.http.app.router.add_get(
-                WSLINK_URL,
-                coordinator.recieved_data if _wslink else unregistred,
-                name="weather_wslink_url",
-            )
-            if debug:
-                _LOGGER.debug("WSLink route: %s", wslink_route)
-
-            routes.add_route(
-                DEFAULT_URL,
-                default_route,
-                coordinator.recieved_data if not _wslink else unregistred,
-                not _wslink,
-            )
-            routes.add_route(
-                WSLINK_URL, wslink_route, coordinator.recieved_data, _wslink
-            )
+        _ = hass.http.app.router.add_get(DEFAULT_URL, routes.dispatch)
+        _ = hass.http.app.router.add_get(WSLINK_URL, routes.dispatch)
 
+        # Save initialised routes
         hass_data["routes"] = routes
 
-        except RuntimeError as Ex:  # pylint: disable=(broad-except)
-            if (
-                "Added route will never be executed, method GET is already registered"
-                in Ex.args
-            ):
-                _LOGGER.info("Handler to URL (%s) already registred", url_path)
-                return False
-
-            _LOGGER.error("Unable to register URL handler! (%s)", Ex.args)
-            return False
-
-        _LOGGER.info(
-            "Registered path to handle weather data: %s",
-            routes.get_enabled(),  # pylint: disable=used-before-assignment
+    except RuntimeError as Ex:
+        _LOGGER.critical(
+            "Routes cannot be added. Integration will not work as expected. %s", Ex
         )
-
-    if _wslink:
-        routes.switch_route(coordinator.recieved_data, WSLINK_URL)
+        raise ConfigEntryNotReady from Ex
     else:
-        routes.switch_route(coordinator.recieved_data, DEFAULT_URL)
-
-    return routes
+        return True
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@@ -212,22 +202,23 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     hass_data = hass.data.setdefault(DOMAIN, {})
     hass_data[entry.entry_id] = coordinator
 
-    _wslink = entry.options.get(WSLINK)
-    debug = entry.options.get(DEV_DBG)
+    routes: Routes | None = hass_data.get("routes", None)
+
+    _wslink = checked_or(entry.options.get(WSLINK), bool, False)
 
-    if debug:
     _LOGGER.debug("WS Link is %s", "enbled" if _wslink else "disabled")
 
-    route = register_path(
-        hass, DEFAULT_URL if not _wslink else WSLINK_URL, coordinator, entry
-    )
+    if routes:
+        _LOGGER.debug("We have routes registered, will try to switch dispatcher.")
+        routes.switch_route(DEFAULT_URL if not _wslink else WSLINK_URL)
+        _LOGGER.debug("%s", routes.show_enabled())
+    else:
+        routes_enabled = register_path(hass, coordinator, entry)
 
-    if not route:
+        if not routes_enabled:
             _LOGGER.error("Fatal: path not registered!")
             raise PlatformNotReady
 
-    hass_data["route"] = route
-
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
     entry.async_on_unload(entry.add_update_listener(update_listener))
@@ -238,7 +229,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 async def update_listener(hass: HomeAssistant, entry: ConfigEntry):
     """Update setup listener."""
 
-    await hass.config_entries.async_reload(entry.entry_id)
+    _ = await hass.config_entries.async_reload(entry.entry_id)
 
     _LOGGER.info("Settings updated")
 

custom_components/sws12500/config_flow.py

@@ -1,18 +1,28 @@
 """Config flow for Sencor SWS 12500 Weather Station integration."""
 
+import secrets
 from typing import Any
 
 import voluptuous as vol
+from yarl import URL
 
-from homeassistant.config_entries import ConfigFlow, ConfigFlowResult, OptionsFlow
+from homeassistant.config_entries import (
+    ConfigEntry,
+    ConfigFlow,
+    ConfigFlowResult,
+    OptionsFlow,
+)
 from homeassistant.core import callback
 from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.network import get_url
 
 from .const import (
     API_ID,
     API_KEY,
     DEV_DBG,
     DOMAIN,
+    ECOWITT_ENABLED,
+    ECOWITT_WEBHOOK_ID,
     INVALID_CREDENTIALS,
     POCASI_CZ_API_ID,
     POCASI_CZ_API_KEY,
@@ -51,10 +61,12 @@ class ConfigOptionsFlowHandler(OptionsFlow):
         self.migrate_schema = {}
         self.pocasi_cz: dict[str, Any] = {}
         self.pocasi_cz_schema = {}
+        self.ecowitt: dict[str, Any] = {}
+        self.ecowitt_schema = {}
 
-        @property
-        def config_entry(self):
-            return self.hass.config_entries.async_get_entry(self.handler)
+        # @property
+        # def config_entry(self) -> ConfigEntry:
+        #     return self.hass.config_entries.async_get_entry(self.handler)
 
     async def _get_entry_data(self):
         """Get entry data."""
@@ -133,15 +145,20 @@ class ConfigOptionsFlowHandler(OptionsFlow):
             ): bool,
         }
 
-    async def async_step_init(self, user_input=None):
+        self.ecowitt = {
+            ECOWITT_WEBHOOK_ID: self.config_entry.options.get(ECOWITT_WEBHOOK_ID, ""),
+            ECOWITT_ENABLED: self.config_entry.options.get(ECOWITT_ENABLED, False),
+        }
+
+    async def async_step_init(self, user_input: dict[str, Any] = {}):
         """Manage the options - show menu first."""
         return self.async_show_menu(
-            step_id="init", menu_options=["basic", "windy", "pocasi"]
+            step_id="init", menu_options=["basic", "ecowitt", "windy", "pocasi"]
         )
 
-    async def async_step_basic(self, user_input=None):
+    async def async_step_basic(self, user_input: Any = None):
         """Manage basic options - credentials."""
-        errors = {}
+        errors: dict[str, str] = {}
 
         await self._get_entry_data()
 
@@ -159,14 +176,7 @@ class ConfigOptionsFlowHandler(OptionsFlow):
         elif user_input[API_KEY] == user_input[API_ID]:
             errors["base"] = "valid_credentials_match"
         else:
-            # retain windy data
-            user_input.update(self.windy_data)
-
-            # retain sensors
-            user_input.update(self.sensors)
-
-            # retain pocasi data
-            user_input.update(self.pocasi_cz)
+            user_input = self.retain_data(user_input)
 
             return self.async_create_entry(title=DOMAIN, data=user_input)
 
@@ -179,9 +189,9 @@ class ConfigOptionsFlowHandler(OptionsFlow):
             errors=errors,
         )
 
-    async def async_step_windy(self, user_input=None):
+    async def async_step_windy(self, user_input: Any = None):
         """Manage windy options."""
-        errors = {}
+        errors: dict[str, str] = {}
 
         await self._get_entry_data()
 
@@ -200,22 +210,14 @@ class ConfigOptionsFlowHandler(OptionsFlow):
                 errors=errors,
             )
 
-        # retain user_data
-        user_input.update(self.user_data)
-
-        # retain senors
-        user_input.update(self.sensors)
-
-        # retain pocasi cz
-
-        user_input.update(self.pocasi_cz)
+        user_input = self.retain_data(user_input)
 
         return self.async_create_entry(title=DOMAIN, data=user_input)
 
     async def async_step_pocasi(self, user_input: Any = None) -> ConfigFlowResult:
         """Handle the pocasi step."""
 
-        errors = {}
+        errors: dict[str, str] = {}
 
         await self._get_entry_data()
 
@@ -241,17 +243,63 @@ class ConfigOptionsFlowHandler(OptionsFlow):
                 data_schema=vol.Schema(self.pocasi_cz_schema),
                 errors=errors,
             )
-        # retain user data
-        user_input.update(self.user_data)
 
-        # retain senors
-        user_input.update(self.sensors)
-
-        # retain windy
-        user_input.update(self.windy_data)
+        user_input = self.retain_data(user_input)
 
         return self.async_create_entry(title=DOMAIN, data=user_input)
 
+    async def async_step_ecowitt(self, user_input: Any = None) -> ConfigFlowResult:
+        """Ecowitt stations setup."""
+
+        errors: dict[str, str] = {}
+        await self._get_entry_data()
+
+        if not (webhook := self.ecowitt.get(ECOWITT_WEBHOOK_ID)):
+            webhook = secrets.token_hex(8)
+
+        if user_input is None:
+            url: URL = URL(get_url(self.hass))
+
+            if not url.host:
+                url.host = "UNKNOWN"
+
+            ecowitt_schema = {
+                vol.Required(
+                    ECOWITT_WEBHOOK_ID,
+                    default=webhook,
+                ): str,
+                vol.Optional(
+                    ECOWITT_ENABLED,
+                    default=self.ecowitt.get(ECOWITT_ENABLED, False),
+                ): bool,
+            }
+
+            return self.async_show_form(
+                step_id="ecowitt",
+                data_schema=vol.Schema(ecowitt_schema),
+                description_placeholders={
+                    "url": url.host,
+                    "port": str(url.port),
+                    "webhook_id": webhook,
+                },
+                errors=errors,
+            )
+
+        user_input = self.retain_data(user_input)
+        return self.async_create_entry(title=DOMAIN, data=user_input)
+
+    def retain_data(self, data: dict[str, Any]) -> dict[str, Any]:
+        """Retain user_data."""
+
+        return {
+            **self.user_data,
+            **self.windy_data,
+            **self.pocasi_cz,
+            **self.sensors,
+            **self.ecowitt,
+            **dict(data),
+        }
+
 
 class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Sencor SWS 12500 Weather Station."""
@@ -265,7 +313,7 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
 
     VERSION = 1
 
-    async def async_step_user(self, user_input=None):
+    async def async_step_user(self, user_input: Any = None):
         """Handle the initial step."""
         if user_input is None:
             await self.async_set_unique_id(DOMAIN)
@@ -276,7 +324,7 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
                 data_schema=vol.Schema(self.data_schema),
             )
 
-        errors = {}
+        errors: dict[str, str] = {}
 
         if user_input[API_ID] in INVALID_CREDENTIALS:
             errors[API_ID] = "valid_credentials_api"
@@ -297,6 +345,6 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
 
     @staticmethod
     @callback
-    def async_get_options_flow(config_entry) -> ConfigOptionsFlowHandler:
+    def async_get_options_flow(config_entry: ConfigEntry) -> ConfigOptionsFlowHandler:
         """Get the options flow for this handler."""
         return ConfigOptionsFlowHandler()

custom_components/sws12500/const.py

@@ -12,6 +12,7 @@ DATABASE_PATH = "/config/home-assistant_v2.db"
 POCASI_CZ_URL: Final = "http://ms.pocasimeteo.cz"
 POCASI_CZ_SEND_MINIMUM: Final = 12  # minimal time to resend data
 
+
 ICON = "mdi:weather"
 
 API_KEY = "API_KEY"
@@ -23,6 +24,10 @@ SENSOR_TO_MIGRATE: Final = "sensor_to_migrate"
 DEV_DBG: Final = "dev_debug_checkbox"
 WSLINK: Final = "wslink"
 
+ECOWITT: Final = "ecowitt"
+ECOWITT_WEBHOOK_ID: Final = "ecowitt_webhook_id"
+ECOWITT_ENABLED: Final = "ecowitt_enabled"
+
 POCASI_CZ_API_KEY = "POCASI_CZ_API_KEY"
 POCASI_CZ_API_ID = "POCASI_CZ_API_ID"
 POCASI_CZ_SEND_INTERVAL = "POCASI_SEND_INTERVAL"

custom_components/sws12500/manifest.json

@@ -8,7 +8,7 @@
   "homekit": {},
   "iot_class": "local_push",
   "issue_tracker": "https://github.com/schizza/SWS-12500-custom-component/issues",
-  "requirements": [],
+  "requirements": ["typecheck-runtime==0.2.0"],
   "ssdp": [],
   "version": "1.6.9",
   "zeroconf": []

custom_components/sws12500/pocasti_cz.py

@@ -5,6 +5,7 @@ import logging
 from typing import Any, Literal
 
 from aiohttp import ClientError
+from py_typecheck.core import checked
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -75,8 +76,20 @@ class PocasiPush:
         """Pushes weather data to server."""
 
         _data = data.copy()
-        _api_id = self.config.options.get(POCASI_CZ_API_ID)
-        _api_key = self.config.options.get(POCASI_CZ_API_KEY)
+
+        if (_api_id := checked(self.config.options.get(POCASI_CZ_API_ID), str)) is None:
+            _LOGGER.error(
+                "No API ID is provided for Pocasi Meteo. Check your configuration."
+            )
+            return
+
+        if (
+            _api_key := checked(self.config.options.get(POCASI_CZ_API_KEY), str)
+        ) is None:
+            _LOGGER.error(
+                "No API Key is provided for Pocasi Meteo. Check your configuration."
+            )
+            return
 
         if self.log:
             _LOGGER.info(
@@ -91,7 +104,7 @@ class PocasiPush:
                 self._interval,
                 self.next_update,
             )
-            return False
+            return
 
         request_url: str = ""
         if mode == "WSLINK":
@@ -139,5 +152,3 @@ class PocasiPush:
 
         if self.log:
             _LOGGER.info("Next update: %s", str(self.next_update))
-
-        return None

custom_components/sws12500/routes.py

@@ -1,77 +1,67 @@
-"""Store routes info."""
+"""Routes implementation."""
 
-from collections.abc import Callable
-from dataclasses import dataclass
-from logging import getLogger
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+import logging
 
-from aiohttp.web import AbstractRoute, Response
+from aiohttp.web import Request, Response
 
-_LOGGER = getLogger(__name__)
+_LOGGER = logging.getLogger(__name__)
+
+Handler = Callable[[Request], Awaitable[Response]]
 
 
 @dataclass
-class Route:
-    """Store route info."""
+class RouteInfo:
+    """Route struct."""
 
     url_path: str
-    route: AbstractRoute
-    handler: Callable
+    handler: Handler
     enabled: bool = False
-
-    def __str__(self):
-        """Return string representation."""
-        return f"{self.url_path} -> {self.handler}"
+    fallback: Handler = field(default_factory=lambda: unregistred)
 
 
 class Routes:
-    """Store routes info."""
+    """Routes class."""
 
     def __init__(self) -> None:
-        """Initialize routes."""
-        self.routes = {}
+        """Init."""
+        self.routes: dict[str, RouteInfo] = {}
 
-    def switch_route(self, coordinator: Callable, url_path: str):
-        """Switch route."""
+    async def dispatch(self, request: Request) -> Response:
+        """Dispatch."""
+        info = self.routes.get(request.path)
+        if not info:
+            _LOGGER.debug("Route %s is not registered!", request.path)
+            return await unregistred(request)
+        handler = info.handler if info.enabled else info.fallback
+        return await handler(request)
 
-        for url, route in self.routes.items():
-            if url == url_path:
-                _LOGGER.info("New coordinator to route: %s", route.url_path)
-                route.enabled = True
-                route.handler = coordinator
-                route.route._handler = coordinator  # noqa: SLF001
-            else:
-                route.enabled = False
-                route.handler = unregistred
-                route.route._handler = unregistred  # noqa: SLF001
+    def switch_route(self, url_path: str) -> None:
+        """Switch route to new handler."""
+        for path, info in self.routes.items():
+            info.enabled = path == url_path
 
     def add_route(
-        self,
-        url_path: str,
-        route: AbstractRoute,
-        handler: Callable,
-        enabled: bool = False,
-    ):
-        """Add route."""
-        self.routes[url_path] = Route(url_path, route, handler, enabled)
+        self, url_path: str, handler: Handler, *, enabled: bool = False
+    ) -> None:
+        """Add route to dispatcher."""
 
-    def get_route(self, url_path: str) -> Route:
-        """Get route."""
-        return self.routes.get(url_path, Route)
+        self.routes[url_path] = RouteInfo(url_path, handler, enabled=enabled)
+        _LOGGER.debug("Registered dispatcher for route %s", url_path)
 
-    def get_enabled(self) -> str:
-        """Get enabled routes."""
-        enabled_routes = [
-            route.url_path for route in self.routes.values() if route.enabled
-        ]
-        return "".join(enabled_routes) if enabled_routes else "None"
-
-    def __str__(self):
-        """Return string representation."""
-        return "\n".join([str(route) for route in self.routes.values()])
+    def show_enabled(self) -> str:
+        """Show info of enabled route."""
+        for url, route in self.routes.items():
+            if route.enabled:
+                return (
+                    f"Dispatcher enabled for URL: {url}, with handler: {route.handler}"
+                )
+        return "No routes is enabled."
 
 
-async def unregistred(*args, **kwargs):
-    """Unregister path to handle incoming data."""
-
-    _LOGGER.error("Recieved data to unregistred webhook. Check your settings")
-    return Response(body=f"{'Unregistred webhook.'}", status=404)
+async def unregistred(request: Request) -> Response:
+    """Return unregistred error."""
+    _ = request
+    _LOGGER.debug("Received data to unregistred or disabled webhook.")
+    return Response(text="Unregistred webhook. Check your settings.", status=400)

custom_components/sws12500/translations/cs.json

@@ -42,6 +42,7 @@
           "basic": "Základní - přístupové údaje (přihlášení)",
           "windy": "Nastavení pro přeposílání dat na Windy",
           "pocasi": "Nastavení pro přeposlání dat na Počasí Meteo CZ",
+          "ecowitt": "Nastavení pro stanice Ecowitt",
           "migration": "Migrace statistiky senzoru"
         }
       },
@@ -92,6 +93,18 @@
           "pocasi_logger_checkbox": "Zapnout pouze v případě, že chcete zaslat ladící informace vývojáři."
         }
       },
+      "ecowitt": {
+        "description": "Nastavení pro Ecowitt",
+        "title": "Konfigurace pro stanice Ecowitt",
+        "data": {
+          "ecowitt_webhook_id": "Unikátní webhook ID",
+          "ecowitt_enabled": "Povolit data ze stanice Ecowitt"
+        },
+        "data_description": {
+          "ecowitt_webhook_id": "Nastavení pro stanici: {url}:{port}/weatherhub/{webhook_id}",
+          "ecowitt_enabled": "Povolit přijímání dat ze stanic Ecowitt"
+        }
+      },
       "migration": {
         "title": "Migrace statistiky senzoru.",
         "description": "Pro správnou funkci dlouhodobé statistiky je nutné provést migraci jednotky senzoru v dlouhodobé statistice. Původní jednotka dlouhodobé statistiky pro denní úhrn srážek byla v mm/d, nicméně stanice zasílá pouze data v mm bez časového rozlišení.\n\n Senzor, který má být migrován je pro denní úhrn srážek. Pokud je v seznamu již správná hodnota u senzoru pro denní úhrn (mm), pak je již migrace hotová.\n\n Výsledek migrace pro senzor: {migration_status}, přepvedeno celkem {migration_count} řádků.",

custom_components/sws12500/translations/en.json

@@ -87,6 +87,18 @@
           "pocasi_logger_checkbox": "Enable only if you want to send debbug data to the developer"
         }
       },
+      "ecowitt": {
+        "description": "Nastavení pro Ecowitt",
+        "title": "Konfigurace pro stanice Ecowitt",
+        "data": {
+          "ecowitt_webhook_id": "Unikátní webhook ID",
+          "ecowitt_enabled": "Povolit data ze stanice Ecowitt"
+        },
+        "data_description": {
+          "ecowitt_webhook_id": "Nastavení pro stanici: {url}:{port}/weatherhub/{webhook_id}",
+          "ecowitt_enabled": "Povolit přijímání dat ze stanic Ecowitt"
+        }
+      },
       "migration": {
         "title": "Statistic migration.",
         "description": "For the correct functioning of long-term statistics, it is necessary to migrate the sensor unit in the long-term statistics. The original unit of long-term statistics for daily precipitation was in mm/d, however, the station only sends data in mm without time differentiation.\n\n The sensor to be migrated is for daily precipitation. If the correct value is already in the list for the daily precipitation sensor (mm), then the migration is already complete.\n\n Migration result for the sensor: {migration_status}, a total of {migration_count} rows converted.",

custom_components/sws12500/utils.py

@@ -2,11 +2,12 @@
 
 import logging
 import math
-from pathlib import Path
-import sqlite3
-from typing import Any
+from multiprocessing import Value
+from typing import Any, cast
 
 import numpy as np
+from py_typecheck import checked
+from py_typecheck.core import checked_or
 
 from homeassistant.components import persistent_notification
 from homeassistant.config_entries import ConfigEntry
@@ -15,7 +16,6 @@ from homeassistant.helpers.translation import async_get_translations
 
 from .const import (
     AZIMUT,
-    DATABASE_PATH,
     DEV_DBG,
     OUTSIDE_HUMIDITY,
     OUTSIDE_TEMP,
@@ -37,19 +37,19 @@ async def translations(
     *,
     key: str = "message",
     category: str = "notify",
-) -> str:
+) -> str | None:
     """Get translated keys for domain."""
 
     localize_key = f"component.{translation_domain}.{category}.{translation_key}.{key}"
 
-    language = hass.config.language
+    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 ""
+    return None
 
 
 async def translated_notification(
@@ -70,7 +70,7 @@ async def translated_notification(
         f"component.{translation_domain}.{category}.{translation_key}.title"
     )
 
-    language = hass.config.language
+    language: str = cast("str", hass.config.language)
 
     _translations = await async_get_translations(
         hass, language, category, [translation_domain]
@@ -91,7 +91,10 @@ async def translated_notification(
 
 
 async def update_options(
-    hass: HomeAssistant, entry: ConfigEntry, update_key, update_value
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    update_key: str,
+    update_value: str | list[str] | bool,
 ) -> bool:
     """Update config.options entry."""
     conf = {**entry.options}
@@ -100,46 +103,43 @@ async def update_options(
     return hass.config_entries.async_update_entry(entry, options=conf)
 
 
-def anonymize(data):
+def anonymize(
+    data: dict[str, str | int | float | bool],
+) -> dict[str, str | int | float | bool]:
     """Anoynimize recieved data."""
-
-    anonym = {}
-    for k in data:
-        if k not in {"ID", "PASSWORD", "wsid", "wspw"}:
-            anonym[k] = data[k]
-
-    return anonym
+    anonym: dict[str, str] = {}
+    return {
+        anonym[key]: value
+        for key, value in data.items()
+        if key not in {"ID", "PASSWORD", "wsid", "wspw"}
+    }
 
 
-def remap_items(entities):
+def remap_items(entities: dict[str, str]) -> dict[str, str]:
     """Remap items in query."""
-    items = {}
-    for item in entities:
-        if item in REMAP_ITEMS:
-            items[REMAP_ITEMS[item]] = entities[item]
-
-    return items
+    return {
+        REMAP_ITEMS[key]: value for key, value in entities.items() if key in REMAP_ITEMS
+    }
 
 
-def remap_wslink_items(entities):
+def remap_wslink_items(entities: dict[str, str]) -> dict[str, str]:
     """Remap items in query for WSLink API."""
-    items = {}
-    for item in entities:
-        if item in REMAP_WSLINK_ITEMS:
-            items[REMAP_WSLINK_ITEMS[item]] = entities[item]
-
-    return items
+    return {
+        REMAP_WSLINK_ITEMS[key]: value
+        for key, value in entities.items()
+        if key in REMAP_WSLINK_ITEMS
+    }
 
 
-def loaded_sensors(config_entry: ConfigEntry) -> list | None:
+def loaded_sensors(config_entry: ConfigEntry) -> list[str]:
     """Get loaded sensors."""
 
     return config_entry.options.get(SENSORS_TO_LOAD) or []
 
 
 def check_disabled(
-    hass: HomeAssistant, items, config_entry: ConfigEntry
-) -> list | None:
+    items: dict[str, str], config_entry: ConfigEntry
+) -> list[str] | None:
     """Check if we have data for unloaded sensors.
 
     If so, then add sensor to load queue.
@@ -147,10 +147,11 @@ def check_disabled(
     Returns list of found sensors or None
     """
 
-    log: bool = config_entry.options.get(DEV_DBG, False)
+    log = checked_or(config_entry.options.get(DEV_DBG), bool, False)
+
     entityFound: bool = False
-    _loaded_sensors = loaded_sensors(config_entry)
-    missing_sensors: list = []
+    _loaded_sensors: list[str] = loaded_sensors(config_entry)
+    missing_sensors: list[str] = []
 
     for item in items:
         if log:
@@ -177,8 +178,8 @@ 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.
+def battery_level(battery: int) -> UnitOfBat:
+    """Return battery level.
 
     Returns UnitOfBat
     """
@@ -188,10 +189,10 @@ def battery_level_to_text(battery: int) -> UnitOfBat:
         1: UnitOfBat.NORMAL,
     }
 
-    if battery is None:
+    if (v := checked(battery, int)) is None:
         return UnitOfBat.UNKNOWN
 
-    return level_map.get(int(battery), UnitOfBat.UNKNOWN)
+    return level_map.get(v, UnitOfBat.UNKNOWN)
 
 
 def battery_level_to_icon(battery: UnitOfBat) -> str:
@@ -218,21 +219,40 @@ def celsius_to_fahrenheit(celsius: float) -> float:
     return celsius * 9.0 / 5.0 + 32
 
 
-def heat_index(data: Any, convert: bool = False) -> float | None:
+def _to_float(val: Any) -> float | None:
+    """Convert int or string to float."""
+
+    if not val:
+        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 recieved data from Celsius to Fahrenheit
     """
-
-    temp = data.get(OUTSIDE_TEMP, None)
-    rh = data.get(OUTSIDE_HUMIDITY, None)
-
-    if not temp or not rh:
+    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
 
-    temp = float(temp)
-    rh = float(rh)
+    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
 
@@ -263,21 +283,30 @@ def heat_index(data: Any, convert: bool = False) -> float | None:
     return simple
 
 
-def chill_index(data: Any, convert: bool = False) -> float | None:
+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 recieved data from Celsius to Fahrenheit
     """
+    temp = _to_float(data.get(OUTSIDE_TEMP))
+    wind = _to_float(data.get(WIND_SPEED))
 
-    temp = data.get(OUTSIDE_TEMP, None)
-    wind = data.get(WIND_SPEED, None)
-
-    if not temp or not wind:
+    if temp is None:
+        _LOGGER.error(
+            "We are missing/invalid OUTSIDE TEMP (%s), cannot calculate wind chill index.",
+            temp,
+        )
         return None
 
-    temp = float(temp)
-    wind = float(wind)
+    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)
@@ -294,109 +323,3 @@ def chill_index(data: Any, convert: bool = False) -> float | None:
         if temp < 50 and wind > 3
         else temp
     )
-
-
-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

custom_components/sws12500/windy_func.py

@@ -2,8 +2,10 @@
 
 from datetime import datetime, timedelta
 import logging
+from typing import Final
 
 from aiohttp.client_exceptions import ClientError
+from py_typecheck.core import checked
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -52,22 +54,22 @@ class WindyPush:
 
     def __init__(self, hass: HomeAssistant, config: ConfigEntry) -> None:
         """Init."""
-        self.hass = hass
-        self.config = config
+        self.hass: Final = hass
+        self.config: Final = 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.last_update: datetime = datetime.now()
+        self.next_update: datetime = datetime.now() + timed(minutes=1)
 
-        self.log = self.config.options.get(WINDY_LOGGER_ENABLED)
-        self.invalid_response_count = 0
+        self.log: bool = self.config.options.get(WINDY_LOGGER_ENABLED, False)
+        self.invalid_response_count: int = 0
 
     def verify_windy_response(  # pylint: disable=useless-return
         self,
         response: str,
-    ) -> WindyNotInserted | WindySuccess | WindyApiKeyError | None:
+    ):
         """Verify answer form Windy."""
 
         if self.log:
@@ -85,9 +87,7 @@ class WindyPush:
         if "Unauthorized" in response:
             raise WindyApiKeyError
 
-        return None
-
-    async def push_data_to_windy(self, data):
+    async def push_data_to_windy(self, data: dict[str, str]) ->  bool:
         """Pushes weather data do Windy stations.
 
         Interval is 5 minutes, otherwise Windy would not accepts data.
@@ -96,8 +96,6 @@ class WindyPush:
         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",
@@ -112,13 +110,18 @@ class WindyPush:
 
         for purge in PURGE_DATA:
             if purge in purged_data:
-                purged_data.pop(purge)
+                _ = 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)
+        if (
+            windy_api_key := checked(self.config.options.get(WINDY_API_KEY), str)
+        ) is None:
+            _LOGGER.error("Windy API key is not provided! Check your configuration.")
+            return False
+
         request_url = f"{WINDY_URL}{windy_api_key}"
 
         if self.log:
@@ -133,27 +136,33 @@ class WindyPush:
                     # 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
 
-                    await update_options(self.hass, self.config, WINDY_ENABLED, False)
+                    if not (
+                        await update_options(
+                            self.hass, self.config, WINDY_ENABLED, False
+                        )
+                    ):
+                        _LOGGER.debug("Failed to set Windy option to false.")
 
                 except WindySuccess:
                     if self.log:
                         _LOGGER.info(WINDY_SUCCESS)
-                    text_for_test = WINDY_SUCCESS
+                else:
+                    if self.log:
+                        _LOGGER.debug(WINDY_NOT_INSERTED)
 
         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
-                await update_options(self.hass, self.config, WINDY_ENABLED, False)
+                if not await update_options(
+                    self.hass, self.config, WINDY_ENABLED, False
+                ):
+                    _LOGGER.debug("Failed to set Windy options to false.")
 
         self.last_update = datetime.now()
         self.next_update = self.last_update + timed(minutes=5)
@@ -161,6 +170,4 @@ class WindyPush:
         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
+        return True