diff --git a/custom_components/bticino_x8000/__init__.py b/custom_components/bticino_x8000/__init__.py index 2731587..69c144c 100644 --- a/custom_components/bticino_x8000/__init__.py +++ b/custom_components/bticino_x8000/__init__.py @@ -11,8 +11,6 @@ from .auth import refresh_access_token from .webhook import BticinoX8000WebhookHandler -# import datetime - PLATFORMS = [Platform.CLIMATE] _LOGGER = logging.getLogger(__name__) @@ -21,11 +19,11 @@ async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, -): +) -> bool: """Set up the Bticino_X8000 component.""" data = dict(config_entry.data) - async def update_token(now): + async def update_token(now: None) -> None: _LOGGER.debug("Refreshing access token") ( access_token, @@ -53,7 +51,7 @@ async def update_token(now): return True -async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry): +async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: """Unload Entry.""" data = config_entry.data bticino_api = BticinoX8000Api(data) diff --git a/custom_components/bticino_x8000/api.py b/custom_components/bticino_x8000/api.py index 1a6e51c..301849c 100644 --- a/custom_components/bticino_x8000/api.py +++ b/custom_components/bticino_x8000/api.py @@ -1,6 +1,7 @@ """Api.""" import json import logging +from typing import Any import aiohttp @@ -19,7 +20,7 @@ class BticinoX8000Api: """Legrand API class.""" - def __init__(self, data) -> None: + def __init__(self, data: dict[str, Any]) -> None: """Init function.""" self.data = data self.header = { @@ -28,7 +29,7 @@ def __init__(self, data) -> None: "Content-Type": "application/json", } - async def check_api_endpoint_health(self): + async def check_api_endpoint_health(self) -> bool: """Check API endpoint helth.""" url = f"{DEFAULT_API_BASE_URL}{AUTH_CHECK_ENDPOINT}" @@ -73,7 +74,7 @@ async def check_api_endpoint_health(self): ) return False - async def handle_unauthorized_error(self, response): + async def handle_unauthorized_error(self, response: aiohttp.ClientResponse) -> bool: """Head off 401 Unauthorized.""" status_code = response.status @@ -89,8 +90,11 @@ async def handle_unauthorized_error(self, response): "Ocp-Apim-Subscription-Key": self.data["subscription_key"], "Content-Type": "application/json", } + return True + else: + return False - async def get_plants(self): + async def get_plants(self) -> dict[str, Any]: """Retrieve thermostat plants.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}{PLANTS}" async with aiohttp.ClientSession() as session: @@ -119,7 +123,7 @@ async def get_plants(self): "error": f"Errore nella richiesta di get_plants: {e}", } - async def get_topology(self, plantId): + async def get_topology(self, plantId: str) -> dict[str, Any]: """Retrieve thermostat topology.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}{PLANTS}/{plantId}{TOPOLOGY}" async with aiohttp.ClientSession() as session: @@ -137,7 +141,7 @@ async def get_topology(self, plantId): # Retry the request on 401 Unauthorized if await self.handle_unauthorized_error(response): # Retry the original request - return await self.get_topology() + return await self.get_topology(plantId) return { "status_code": status_code, "error": f"Failed to get topology: Content: {content}, HEADEr: {self.header}, URL: {url}", @@ -148,7 +152,9 @@ async def get_topology(self, plantId): "error": f"Failed to get topology: {e}", } - async def set_chronothermostat_status(self, plantId, moduleId, data): + async def set_chronothermostat_status( + self, plantId: str, moduleId: str, data: dict[str, Any] + ) -> dict[str, Any]: """Set thermostat status.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}/chronothermostat/thermoregulation/addressLocation{PLANTS}/{plantId}/modules/parameter/id/value/{moduleId}" async with aiohttp.ClientSession() as session: @@ -173,7 +179,9 @@ async def set_chronothermostat_status(self, plantId, moduleId, data): "error": f"Errore nella richiesta di set_chronothermostat_status: {e}", } - async def get_chronothermostat_status(self, plantId, moduleId): + async def get_chronothermostat_status( + self, plantId: str, moduleId: str + ) -> dict[str, Any]: """Get thermostat status.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}/chronothermostat/thermoregulation/addressLocation{PLANTS}/{plantId}/modules/parameter/id/value/{moduleId}" async with aiohttp.ClientSession() as session: @@ -194,7 +202,9 @@ async def get_chronothermostat_status(self, plantId, moduleId): "error": f"Errore nella richiesta di get_chronothermostat_status: {e}", } - async def get_chronothermostat_measures(self, plantId, moduleId): + async def get_chronothermostat_measures( + self, plantId: str, moduleId: str + ) -> dict[str, Any]: """Get thermostat measures.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}/chronothermostat/thermoregulation/addressLocation{PLANTS}/{plantId}/modules/parameter/id/value/{moduleId}/measures" async with aiohttp.ClientSession() as session: @@ -217,7 +227,9 @@ async def get_chronothermostat_measures(self, plantId, moduleId): "error": f"Errore nella richiesta di get_chronothermostat_measures: {e}", } - async def get_chronothermostat_programlist(self, plantId, moduleId): + async def get_chronothermostat_programlist( + self, plantId: str, moduleId: str + ) -> dict[str, Any]: """Get thermostat programlist.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}/chronothermostat/thermoregulation/addressLocation{PLANTS}/{plantId}/modules/parameter/id/value/{moduleId}/programlist" async with aiohttp.ClientSession() as session: @@ -243,7 +255,7 @@ async def get_chronothermostat_programlist(self, plantId, moduleId): "error": f"Errore nella richiesta di get_chronothermostat_programlist: {e}", } - async def get_subscriptions_C2C_notifications(self): + async def get_subscriptions_C2C_notifications(self) -> dict[str, Any]: """Get C2C subscriptions.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}/subscription" async with aiohttp.ClientSession() as session: @@ -267,7 +279,9 @@ async def get_subscriptions_C2C_notifications(self): "error": f"Errore nella richiesta di get_subscriptions_C2C_notifications: {e}", } - async def set_subscribe_C2C_notifications(self, plantId, data): + async def set_subscribe_C2C_notifications( + self, plantId: str, data: dict[str, Any] + ) -> dict[str, Any]: """Add C2C subscriptions.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}{PLANTS}/{plantId}/subscription" async with aiohttp.ClientSession() as session: @@ -290,7 +304,9 @@ async def set_subscribe_C2C_notifications(self, plantId, data): "error": f"Errore nella richiesta di set_subscribe_C2C_notifications: {e}", } - async def delete_subscribe_C2C_notifications(self, plantId, subscriptionId): + async def delete_subscribe_C2C_notifications( + self, plantId: str, subscriptionId: str + ) -> dict[str, Any]: """Remove C2C subscriptions.""" url = f"{DEFAULT_API_BASE_URL}{THERMOSTAT_API_ENDPOINT}{PLANTS}/{plantId}/subscription/{subscriptionId}" async with aiohttp.ClientSession() as session: diff --git a/custom_components/bticino_x8000/auth.py b/custom_components/bticino_x8000/auth.py index bd7bb97..47f0e14 100644 --- a/custom_components/bticino_x8000/auth.py +++ b/custom_components/bticino_x8000/auth.py @@ -1,5 +1,6 @@ from datetime import timedelta # noqa: D100 import logging +from typing import Any import aiohttp @@ -10,7 +11,9 @@ _LOGGER = logging.getLogger(__name__) -async def exchange_code_for_tokens(client_id, client_secret, redirect_uri, code): +async def exchange_code_for_tokens( + client_id: str, client_secret: str, redirect_uri: str, code: str +) -> tuple[str, str, str]: """Get access token.""" token_url = f"{DEFAULT_AUTH_BASE_URL}{AUTH_REQ_ENDPOINT}" payload = { @@ -34,7 +37,7 @@ async def exchange_code_for_tokens(client_id, client_secret, redirect_uri, code) return access_token, refresh_token, access_token_expires_on -async def refresh_access_token(data): +async def refresh_access_token(data: dict[str, Any]) -> tuple[str, str, str]: """Refresh access token.""" token_url = f"{DEFAULT_AUTH_BASE_URL}{AUTH_REQ_ENDPOINT}" payload = { diff --git a/custom_components/bticino_x8000/config_flow.py b/custom_components/bticino_x8000/config_flow.py index 0a6992b..cbc036f 100644 --- a/custom_components/bticino_x8000/config_flow.py +++ b/custom_components/bticino_x8000/config_flow.py @@ -1,12 +1,14 @@ """Config Flow.""" import logging import secrets +from typing import Any from urllib.parse import parse_qs, urlparse import voluptuous as vol from homeassistant import config_entries from homeassistant.components.webhook import async_generate_id as generate_id +from homeassistant.data_entry_flow import FlowResult from homeassistant.helpers import config_validation as cv from .api import BticinoX8000Api @@ -24,10 +26,12 @@ _LOGGER = logging.getLogger(__name__) -class BticinoX8000ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): +class BticinoX8000ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): # type:ignore """Bticino ConfigFlow.""" - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """User configuration.""" try: external_url = self.hass.config.external_url @@ -87,7 +91,7 @@ async def async_step_user(self, user_input=None): errors={"base": message}, ) - def get_authorization_url(self, user_input): + def get_authorization_url(self, user_input: dict[str, Any]) -> str: """Compose the auth url.""" state = secrets.token_hex(16) return ( @@ -98,7 +102,9 @@ def get_authorization_url(self, user_input): + f"&redirect_uri={DEFAULT_REDIRECT_URI}" ) - async def async_step_get_authorize_code(self, user_input=None): + async def async_step_get_authorize_code( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Get authorization code.""" if user_input is not None: try: @@ -185,7 +191,7 @@ async def async_step_get_authorize_code(self, user_input=None): return await self.async_step_get_authorize_code() return await self.async_step_user(self.data) - async def add_c2c_subscription(self, plantId, webhook_id): + async def add_c2c_subscription(self, plantId: str, webhook_id: str) -> str | None: """Subscribe C2C.""" webhook_path = "/api/webhook/" webhook_endpoint = self.data["external_url"] + webhook_path + webhook_id @@ -194,19 +200,27 @@ async def add_c2c_subscription(self, plantId, webhook_id): ) if response["status_code"] == 201: _LOGGER.debug("Webhook subscription registrata con successo!") - return response["text"]["subscriptionId"] - - async def get_programs_from_api(self, plant_id, topology_id): + subscriptionId: str = response["text"]["subscriptionId"] + return subscriptionId + else: + return None + + async def get_programs_from_api( + self, plant_id: str, topology_id: str + ) -> list[dict[str, Any]]: """Retreive the program list.""" programs = await self.bticino_api.get_chronothermostat_programlist( plant_id, topology_id ) - for program in programs["data"]: - if program["number"] == 0: - programs["data"].remove(program) - return programs["data"] + filtered_programs = [ + program for program in programs["data"] if program["number"] != 0 + ] + + return filtered_programs - async def async_step_select_thermostats(self, user_input): + async def async_step_select_thermostats( + self, user_input: dict[str, Any] + ) -> FlowResult: """User can select one o more thermostat to add.""" selected_thermostats = [ { diff --git a/custom_components/bticino_x8000/webhook.py b/custom_components/bticino_x8000/webhook.py index 1c2bb18..01840fb 100644 --- a/custom_components/bticino_x8000/webhook.py +++ b/custom_components/bticino_x8000/webhook.py @@ -20,13 +20,15 @@ class BticinoX8000WebhookHandler: def __init__( self, hass: HomeAssistant, - webhook_id, + webhook_id: str, ) -> None: """Init.""" self.hass = hass - self.webhook_id = webhook_id + self.webhook_id: str = webhook_id - async def handle_webhook(self, hass: HomeAssistant, webhook_id, request) -> None: + async def handle_webhook( + self, hass: HomeAssistant, webhook_id: str, request: Request + ) -> Response: """Handle webhook.""" try: data = await request.json() @@ -39,7 +41,7 @@ async def handle_webhook(self, hass: HomeAssistant, webhook_id, request) -> None async_dispatcher_send(hass, f"{DOMAIN}_webhook_update", {"data": data}) return Response(text="OK", status=200) - async def async_register_webhook(self): + async def async_register_webhook(self) -> None: """Register the webhook.""" webhook_register( self.hass, @@ -50,7 +52,7 @@ async def async_register_webhook(self): local_only=False, ) - async def async_remove_webhook(self): + async def async_remove_webhook(self) -> None: """Remove the webhook.""" _LOGGER.debug("Unregister webhook with id: %s ", self.webhook_id) webhook_unregister(self.hass, self.webhook_id) diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 0000000..c563eb1 --- /dev/null +++ b/mypy.ini @@ -0,0 +1,19 @@ +[mypy] +python_version = 3.11 +show_error_codes = true +follow_imports = silent +ignore_missing_imports = true +strict_equality = true +warn_incomplete_stub = true +warn_redundant_casts = true +warn_unused_configs = true +warn_unused_ignores = true +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +no_implicit_optional = true +warn_return_any = true +warn_unreachable = true \ No newline at end of file