From fa7e94bcc2ee4976b6d9691ac3d842d5aab115f8 Mon Sep 17 00:00:00 2001 From: Robbe-B <103053873+Robbe-B@users.noreply.github.com> Date: Sun, 12 Jan 2025 13:24:27 +0100 Subject: [PATCH] Resolve some pylinter stuff in the _init_.py --- custom_components/maestro_mcz/__init__.py | 201 +++++++++++++++------- 1 file changed, 135 insertions(+), 66 deletions(-) diff --git a/custom_components/maestro_mcz/__init__.py b/custom_components/maestro_mcz/__init__.py index 0f60738..f280371 100644 --- a/custom_components/maestro_mcz/__init__.py +++ b/custom_components/maestro_mcz/__init__.py @@ -1,33 +1,42 @@ """The maestro_mcz integration.""" + from __future__ import annotations + +import asyncio from datetime import timedelta import logging import os -import asyncio -import async_timeout -from .config_flow import CONF_POLLING_INTERVAL -from .maestro.responses.model import Configuration, ModelConfiguration, SensorConfiguration, SensorConfigurationMultipleModes -from .models import MczConfigItem from homeassistant.config_entries import ConfigEntry -from homeassistant.const import Platform +from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform from homeassistant.core import HomeAssistant - from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.update_coordinator import DataUpdateCoordinator -from homeassistant.const import ( - CONF_USERNAME, - CONF_PASSWORD, -) -from .const import DOMAIN, MANUFACTURER, DEFAULT_POLLING_INTERVAL +from .config_flow import CONF_POLLING_INTERVAL +from .const import DEFAULT_POLLING_INTERVAL, DOMAIN, MANUFACTURER from .maestro import MaestroStove from .maestro.controller.controller_interface import MaestroControllerInterface from .maestro.controller.maestro_controller import MaestroController from .maestro.controller.mocked_controller import MockedController +from .maestro.responses.model import ( + ModelConfiguration, + SensorConfiguration, + SensorConfigurationMultipleModes, +) +from .models import MczConfigItem - -PLATFORMS: list[Platform] = [Platform.CLIMATE, Platform.SENSOR, Platform.FAN, Platform.NUMBER, Platform.SWITCH, Platform.SELECT, Platform.BINARY_SENSOR, Platform.BUTTON, Platform.DATETIME] +PLATFORMS: list[Platform] = [ + Platform.BINARY_SENSOR, + Platform.BUTTON, + Platform.CLIMATE, + Platform.DATETIME, + Platform.FAN, + Platform.NUMBER, + Platform.SELECT, + Platform.SENSOR, + Platform.SWITCH, +] _LOGGER = logging.getLogger(__name__) @@ -36,7 +45,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: hass.data.setdefault(DOMAIN, {}) - pollling_interval = entry.options.get(CONF_POLLING_INTERVAL, DEFAULT_POLLING_INTERVAL) + pollling_interval = entry.options.get( + CONF_POLLING_INTERVAL, DEFAULT_POLLING_INTERVAL + ) mocked_files = await has_mocked_files() @@ -45,27 +56,26 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD] ) await maestroapi.Login() - else: + else: maestroapi: MaestroControllerInterface = MockedController(mocked_files) await maestroapi.Login() - stoveList = [] for stove in maestroapi.Stoves: - stove:MaestroStove = stove coordinator = MczCoordinator(hass, stove, pollling_interval) await coordinator.async_config_entry_first_refresh() - if(coordinator.maestroapi.UniqueCode): #avoid adding a stove without serial number as this is the unique identifier in HA + if coordinator.maestroapi.UniqueCode: # avoid adding a stove without serial number as this is the unique identifier in HA stoveList.append(coordinator) hass.data[DOMAIN][entry.entry_id] = stoveList entry.async_on_unload(entry.add_update_listener(_async_update_listener)) - + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True + async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): @@ -73,21 +83,24 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return unload_ok + async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: """Handle options update.""" await hass.config_entries.async_reload(entry.entry_id) + async def has_mocked_files() -> list[str] | None: + """Check if there are mocked files, if so returns it.""" try: loop = asyncio.get_running_loop() folder_path = "config/custom_components/maestro_mcz/mocked" files_in_dir = await loop.run_in_executor(None, os.listdir, folder_path) - if(files_in_dir is not None): + if files_in_dir is not None: return [folder_path + "/" + file for file in files_in_dir] - else: - return None except: return None + else: + return None class MczCoordinator(DataUpdateCoordinator): @@ -95,7 +108,12 @@ class MczCoordinator(DataUpdateCoordinator): _avoid_ping: bool = False - def __init__(self, hass, maestroapi, polling_interval): + def __init__( + self, + hass: HomeAssistant | None, + maestroapi: MaestroStove, + polling_interval: int, + ) -> None: """Initialize my coordinator.""" super().__init__( hass, @@ -103,18 +121,18 @@ def __init__(self, hass, maestroapi, polling_interval): name="MCZ Stove", update_interval=timedelta(seconds=polling_interval), ) - self._maestroapi:MaestroStove = maestroapi + self._maestroapi = maestroapi async def _async_update_data(self): """Fetch data from API endpoint.""" - async with async_timeout.timeout(15): + async with asyncio.timeout(10): await self._maestroapi.Refresh(not self._avoid_ping) return True - + async def update_data_after_set(self): - """force refresh of data from API endpoint after a SET was executed.""" - #we need to wait here because there is an actual delay between sending a SET and receiving the updated value from the polled MCZ database - await asyncio.sleep(3) + """Force refresh of data from API endpoint after a SET was executed.""" + # we need to wait here because there is an actual delay between sending a SET and receiving the updated value from the polled MCZ database + await asyncio.sleep(3) await self.async_refresh() await asyncio.sleep(3) self._avoid_ping = True @@ -123,72 +141,123 @@ async def update_data_after_set(self): @property def maestroapi(self) -> MaestroStove: + """Return the Maestro api.""" return self._maestroapi - + def get_device_info(self) -> DeviceInfo: + """Return device info.""" sw_version = "" - if(self.maestroapi.Status.is_connected == True): - sw_version=(f"{self._maestroapi.Status.sm_nome_app}.{self._maestroapi.Status.sm_vs_app}"+ f", Panel:{self._maestroapi.Status.mc_vs_app}"+ f", DB:{self._maestroapi.Status.nome_banca_dati_sel}") + if self.maestroapi.Status.is_connected: + sw_version = ( + f"{self._maestroapi.Status.sm_nome_app}.{self._maestroapi.Status.sm_vs_app}" + f", Panel:{self._maestroapi.Status.mc_vs_app}" + f", DB:{self._maestroapi.Status.nome_banca_dati_sel}" + ) else: - sw_version="Device Disconnected" + sw_version = "Device Disconnected" return DeviceInfo( identifiers={(DOMAIN, self._maestroapi.UniqueCode)}, name=self._maestroapi.Name, manufacturer=MANUFACTURER, model=self._maestroapi.Model.model_name, - sw_version=sw_version + sw_version=sw_version, + ) + + def get_model_configuration_by_model_configuration_name( + self, model_configuration_name: str + ) -> ModelConfiguration | None: + """Get the model configuration by the name of the model configuration.""" + return next( + ( + x + for x in self._maestroapi.Model.model_configurations + if x.configuration_name is not None + and model_configuration_name is not None + and x.configuration_name.lower().strip() + == model_configuration_name.lower().strip() + ), + None, + ) + + def get_sensor_configuration_by_model_configuration_name_and_sensor_name( + self, model_configuration_name: str, sensor_name: str + ) -> SensorConfiguration | None: + """Get the sensor configuration by the name of the model configuration and sensor name.""" + model_configuration = self.get_model_configuration_by_model_configuration_name( + model_configuration_name ) - - def get_model_configuration_by_model_configuration_name(self, model_configuration_name:str) -> ModelConfiguration | None: - return next((x for x in self._maestroapi.Model.model_configurations if x.configuration_name is not None and model_configuration_name is not None and x.configuration_name.lower().strip() == model_configuration_name.lower().strip()), None) - - - def get_sensor_configuration_by_model_configuration_name_and_sensor_name(self, model_configuration_name:str, sensor_name:str) -> SensorConfiguration | None: - model_configuration = self.get_model_configuration_by_model_configuration_name(model_configuration_name) - if(model_configuration is None): + if model_configuration is None: return None - else: - sensor_configuration = next((x for x in model_configuration.configurations if x.sensor_name is not None and sensor_name is not None and x.sensor_name.lower() == sensor_name.lower()), None) - if(sensor_configuration is not None): - return SensorConfiguration(sensor_configuration, model_configuration.configuration_id) - - def get_first_matching_sensor_configuration_by_model_configuration_name_and_sensor_name(self, mcz_config_items_list_to_match: list[MczConfigItem]) -> tuple[MczConfigItem,SensorConfiguration] | None: + sensor_configuration = next( + ( + x + for x in model_configuration.configurations + if x.sensor_name is not None + and sensor_name is not None + and x.sensor_name.lower() == sensor_name.lower() + ), + None, + ) + if sensor_configuration is not None: + return SensorConfiguration( + sensor_configuration, model_configuration.configuration_id + ) + return None + + def get_first_matching_sensor_configuration_by_model_configuration_name_and_sensor_name( + self, mcz_config_items_list_to_match: list[MczConfigItem] + ) -> tuple[MczConfigItem, SensorConfiguration] | None: + """Get the first sensor configuration matching by the name of the model configuration and sensor name.""" for x in mcz_config_items_list_to_match: if x.sensor_set_config_name is not None: - matching_configuration = self.get_sensor_configuration_by_model_configuration_name_and_sensor_name(x.sensor_set_config_name, x.sensor_set_name) - if(matching_configuration is not None): + matching_configuration = self.get_sensor_configuration_by_model_configuration_name_and_sensor_name( + x.sensor_set_config_name, x.sensor_set_name + ) + if matching_configuration is not None: return (x, matching_configuration) return None - - def get_all_matching_sensor_configurations_by_model_configuration_name_and_sensor_name(self, mcz_config_items_list_to_match: list[MczConfigItem]) -> list(tuple[MczConfigItem,SensorConfiguration]) | None: + + def get_all_matching_sensor_configurations_by_model_configuration_name_and_sensor_name( + self, mcz_config_items_list_to_match: list[MczConfigItem] + ) -> list[tuple[MczConfigItem, SensorConfiguration]] | None: + """Get the all sensor configurations matching by the name of the model configuration and sensor name.""" temp_list = [] for x in mcz_config_items_list_to_match: if x.sensor_set_config_name is not None: - matching_configuration = self.get_sensor_configuration_by_model_configuration_name_and_sensor_name(x.sensor_set_config_name, x.sensor_set_name) - if(matching_configuration is not None): + matching_configuration = self.get_sensor_configuration_by_model_configuration_name_and_sensor_name( + x.sensor_set_config_name, x.sensor_set_name + ) + if matching_configuration is not None: temp_list.append((x, matching_configuration)) if temp_list: return temp_list - else: - return None - + return None - def get_all_matching_sensor_for_all_configurations_by_model_mode_and_sensor_name(self, mcz_config_items_list_to_match: list[MczConfigItem]) -> list(tuple[MczConfigItem,SensorConfigurationMultipleModes]) | None: + def get_all_matching_sensor_for_all_configurations_by_model_mode_and_sensor_name( + self, mcz_config_items_list_to_match: list[MczConfigItem] + ) -> list[tuple[MczConfigItem, SensorConfigurationMultipleModes]] | None: + """Get the all sensor configurations matching for the matching mode by the name of the model configuration and sensor name.""" temp_list = [] for x in mcz_config_items_list_to_match: if x.mode_to_configuration_name_mapping is not None: - temp_mode_configurations: dict[str,SensorConfiguration] = {} + temp_mode_configurations: dict[str, SensorConfiguration] = {} for mode in x.mode_to_configuration_name_mapping: - matching_configuration = self.get_sensor_configuration_by_model_configuration_name_and_sensor_name(x.mode_to_configuration_name_mapping[mode], x.sensor_set_name) - if(matching_configuration is not None): + matching_configuration = self.get_sensor_configuration_by_model_configuration_name_and_sensor_name( + x.mode_to_configuration_name_mapping[mode], x.sensor_set_name + ) + if matching_configuration is not None: temp_mode_configurations[mode] = matching_configuration - if(temp_mode_configurations is not None and len(temp_mode_configurations) > 0): - temp_list.append((x,SensorConfigurationMultipleModes(temp_mode_configurations))) + if ( + temp_mode_configurations is not None + and len(temp_mode_configurations) > 0 + ): + temp_list.append( + (x, SensorConfigurationMultipleModes(temp_mode_configurations)) + ) if temp_list: return temp_list - else: - return None + return None