add some changes for mypy

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)

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:

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 = {

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 = [
             {

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)

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