Code quality cleanup

custom_components/solaredge_modbus_multi/__init__.py

@@ -1,21 +1,34 @@
 """The SolarEdge Modbus Integration."""
 import logging
 from datetime import timedelta
+from typing import Any
 
 import async_timeout
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import (CONF_HOST, CONF_NAME, CONF_PORT,
-                                 CONF_SCAN_INTERVAL, Platform)
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_NAME,
+    CONF_PORT,
+    CONF_SCAN_INTERVAL,
+    Platform,
+)
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.update_coordinator import (DataUpdateCoordinator,
-                                                      UpdateFailed)
-
-from .const import (CONF_DETECT_BATTERIES, CONF_DETECT_METERS, CONF_DEVICE_ID,
-                    CONF_KEEP_MODBUS_OPEN, CONF_NUMBER_INVERTERS,
-                    CONF_SINGLE_DEVICE_ENTITY, DEFAULT_DETECT_BATTERIES,
-                    DEFAULT_DETECT_METERS, DEFAULT_KEEP_MODBUS_OPEN,
-                    DEFAULT_SCAN_INTERVAL, DEFAULT_SINGLE_DEVICE_ENTITY,
-                    DOMAIN)
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    CONF_DETECT_BATTERIES,
+    CONF_DETECT_METERS,
+    CONF_DEVICE_ID,
+    CONF_KEEP_MODBUS_OPEN,
+    CONF_NUMBER_INVERTERS,
+    CONF_SINGLE_DEVICE_ENTITY,
+    DEFAULT_DETECT_BATTERIES,
+    DEFAULT_DETECT_METERS,
+    DEFAULT_KEEP_MODBUS_OPEN,
+    DEFAULT_SCAN_INTERVAL,
+    DEFAULT_SINGLE_DEVICE_ENTITY,
+    DOMAIN,
+)
 from .hub import DataUpdateFailed, HubInitFailed, SolarEdgeModbusMultiHub
 
 _LOGGER = logging.getLogger(__name__)
@@ -25,7 +38,7 @@
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up SolarEdge Modbus from a config entry."""
-    
+
     entry_updates: dict[str, Any] = {}
     if CONF_SCAN_INTERVAL in entry.data:
         data = {**entry.data}
@@ -36,7 +49,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         }
     if entry_updates:
         hass.config_entries.async_update_entry(entry, **entry_updates)
-    
+
     solaredge_hub = SolarEdgeModbusMultiHub(
         hass,
         entry.data[CONF_NAME],
@@ -49,63 +62,66 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         entry.options.get(CONF_SINGLE_DEVICE_ENTITY, DEFAULT_SINGLE_DEVICE_ENTITY),
         entry.options.get(CONF_KEEP_MODBUS_OPEN, DEFAULT_KEEP_MODBUS_OPEN),
     )
-    
+
     coordinator = SolarEdgeCoordinator(
         hass,
         solaredge_hub,
-        entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
+        entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
     )
-    
+
     hass.data.setdefault(DOMAIN, {})
     hass.data[DOMAIN][entry.entry_id] = {
         "hub": solaredge_hub,
         "coordinator": coordinator,
     }
-    
+
     await coordinator.async_config_entry_first_refresh()
-    
+
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
-    
+
     entry.async_on_unload(entry.add_update_listener(async_reload_entry))
-    
+
     return True
 
+
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
-    solaredge_hub = hass.data[DOMAIN][entry.entry_id]['hub']  
+    solaredge_hub = hass.data[DOMAIN][entry.entry_id]["hub"]
     unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
-    
+
     if unload_ok:
         await solaredge_hub.shutdown()
         hass.data[DOMAIN].pop(entry.entry_id)
-    
+
     return unload_ok
 
+
 async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
     """Handle an options update."""
     await hass.config_entries.async_reload(entry.entry_id)
 
+
 class SolarEdgeCoordinator(DataUpdateCoordinator):
     def __init__(self, hass, hub, scan_interval):
         super().__init__(
             hass,
             _LOGGER,
-            name = "SolarEdge Coordinator",
-            update_interval = timedelta(seconds=scan_interval),
+            name="SolarEdge Coordinator",
+            update_interval=timedelta(seconds=scan_interval),
         )
         self._hub = hub
-        
+
         if scan_interval < 10:
             _LOGGER.warning("Polling frequency < 10, requiring keep modbus open.")
             self._hub.keep_modbus_open = True
-    
+
     async def _async_update_data(self):
         try:
             async with async_timeout.timeout(5):
                 return await self._hub.async_refresh_modbus_data()
-        
+
         except HubInitFailed as e:
             raise UpdateFailed(f"{e}")
-        
+
         except DataUpdateFailed as e:
             raise UpdateFailed(f"{e}")

custom_components/solaredge_modbus_multi/config_flow.py

@@ -5,19 +5,28 @@
 import voluptuous as vol
 from homeassistant import config_entries
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import (CONF_HOST, CONF_NAME, CONF_PORT,
-                                 CONF_SCAN_INTERVAL)
+from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, CONF_SCAN_INTERVAL
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import FlowResult
 
-from .const import (CONF_DETECT_BATTERIES, CONF_DETECT_METERS, CONF_DEVICE_ID,
-                    CONF_KEEP_MODBUS_OPEN, CONF_NUMBER_INVERTERS,
-                    CONF_SINGLE_DEVICE_ENTITY, DEFAULT_DETECT_BATTERIES,
-                    DEFAULT_DETECT_METERS, DEFAULT_DEVICE_ID,
-                    DEFAULT_KEEP_MODBUS_OPEN, DEFAULT_NAME,
-                    DEFAULT_NUMBER_INVERTERS, DEFAULT_PORT,
-                    DEFAULT_SCAN_INTERVAL, DEFAULT_SINGLE_DEVICE_ENTITY,
-                    DOMAIN)
+from .const import (
+    CONF_DETECT_BATTERIES,
+    CONF_DETECT_METERS,
+    CONF_DEVICE_ID,
+    CONF_KEEP_MODBUS_OPEN,
+    CONF_NUMBER_INVERTERS,
+    CONF_SINGLE_DEVICE_ENTITY,
+    DEFAULT_DETECT_BATTERIES,
+    DEFAULT_DETECT_METERS,
+    DEFAULT_DEVICE_ID,
+    DEFAULT_KEEP_MODBUS_OPEN,
+    DEFAULT_NAME,
+    DEFAULT_NUMBER_INVERTERS,
+    DEFAULT_PORT,
+    DEFAULT_SCAN_INTERVAL,
+    DEFAULT_SINGLE_DEVICE_ENTITY,
+    DOMAIN,
+)
 
 
 def host_valid(host):
@@ -29,13 +38,15 @@ def host_valid(host):
         disallowed = re.compile(r"[^a-zA-Z\d\-]")
         return all(x and not disallowed.search(x) for x in host.split("."))
 
+
 @callback
 def solaredge_modbus_multi_entries(hass: HomeAssistant):
     """Return the hosts already configured."""
     return set(
         entry.data[CONF_HOST] for entry in hass.config_entries.async_entries(DOMAIN)
     )
 
+
 class SolaredgeModbusMultiConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     """Solaredge Modbus configflow."""
 
@@ -58,7 +69,7 @@ async def async_step_user(self, user_input=None):
         errors = {}
 
         if user_input is not None:
-            
+
             if self._host_in_configuration_exists(user_input[CONF_HOST]):
                 errors[CONF_HOST] = "already_configured"
             elif not host_valid(user_input[CONF_HOST]):
@@ -68,11 +79,11 @@ async def async_step_user(self, user_input=None):
             elif user_input[CONF_PORT] > 65535:
                 errors[CONF_PORT] = "invalid_tcp_port"
             elif user_input[CONF_DEVICE_ID] > 247:
-                errors[CONF_DEVICE_ID] = "max_device_id"  
+                errors[CONF_DEVICE_ID] = "max_device_id"
             elif user_input[CONF_DEVICE_ID] < 1:
                 errors[CONF_DEVICE_ID] = "min_device_id"
             elif user_input[CONF_NUMBER_INVERTERS] > 32:
-                errors[CONF_NUMBER_INVERTERS] = "max_inverters"  
+                errors[CONF_NUMBER_INVERTERS] = "max_inverters"
             elif user_input[CONF_NUMBER_INVERTERS] < 1:
                 errors[CONF_NUMBER_INVERTERS] = "min_inverters"
             elif user_input[CONF_NUMBER_INVERTERS] + user_input[CONF_DEVICE_ID] > 247:
@@ -96,75 +107,85 @@ async def async_step_user(self, user_input=None):
             step_id="user",
             data_schema=vol.Schema(
                 {
-                    vol.Optional(
-                        CONF_NAME, default=user_input[CONF_NAME]
-                    ): cv.string,
-                    vol.Required(
-                        CONF_HOST, default=user_input[CONF_HOST]
-                    ): cv.string,
-                    vol.Required(
-                        CONF_PORT, default=user_input[CONF_PORT]
-                    ): vol.Coerce(int),
+                    vol.Optional(CONF_NAME, default=user_input[CONF_NAME]): cv.string,
+                    vol.Required(CONF_HOST, default=user_input[CONF_HOST]): cv.string,
+                    vol.Required(CONF_PORT, default=user_input[CONF_PORT]): vol.Coerce(
+                        int
+                    ),
                     vol.Required(
-                        CONF_NUMBER_INVERTERS, default=user_input[CONF_NUMBER_INVERTERS]
+                        CONF_NUMBER_INVERTERS,
+                        default=user_input[CONF_NUMBER_INVERTERS],
                     ): vol.Coerce(int),
                     vol.Required(
                         CONF_DEVICE_ID, default=user_input[CONF_DEVICE_ID]
                     ): vol.Coerce(int),
                 },
             ),
-            errors = errors
+            errors=errors,
         )
 
+
 class SolaredgeModbusMultiOptionsFlowHandler(config_entries.OptionsFlow):
     def __init__(self, config_entry: ConfigEntry):
         """Initialize options flow."""
         self.config_entry = config_entry
 
-    async def async_step_init(self, user_input = None) -> FlowResult:
+    async def async_step_init(self, user_input=None) -> FlowResult:
         errors = {}
 
         """Manage the options."""
         if user_input is not None:
-            
+
             if user_input[CONF_SCAN_INTERVAL] < 1:
                 errors[CONF_SCAN_INTERVAL] = "invalid_scan_interval"
             elif user_input[CONF_SCAN_INTERVAL] > 86400:
                 errors[CONF_SCAN_INTERVAL] = "invalid_scan_interval"
             else:
-                return self.async_create_entry(
-                    title = "",
-                    data = user_input
-                )
+                return self.async_create_entry(title="", data=user_input)
         else:
             user_input = {
-                CONF_SCAN_INTERVAL: self.config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
-                CONF_SINGLE_DEVICE_ENTITY: self.config_entry.options.get(CONF_SINGLE_DEVICE_ENTITY, DEFAULT_SINGLE_DEVICE_ENTITY),
-                CONF_KEEP_MODBUS_OPEN: self.config_entry.options.get(CONF_KEEP_MODBUS_OPEN, DEFAULT_KEEP_MODBUS_OPEN),
-                CONF_DETECT_METERS: self.config_entry.options.get(CONF_DETECT_METERS, DEFAULT_DETECT_METERS),
-                CONF_DETECT_BATTERIES: self.config_entry.options.get(CONF_DETECT_BATTERIES, DEFAULT_DETECT_BATTERIES),
+                CONF_SCAN_INTERVAL: self.config_entry.options.get(
+                    CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
+                ),
+                CONF_SINGLE_DEVICE_ENTITY: self.config_entry.options.get(
+                    CONF_SINGLE_DEVICE_ENTITY, DEFAULT_SINGLE_DEVICE_ENTITY
+                ),
+                CONF_KEEP_MODBUS_OPEN: self.config_entry.options.get(
+                    CONF_KEEP_MODBUS_OPEN, DEFAULT_KEEP_MODBUS_OPEN
+                ),
+                CONF_DETECT_METERS: self.config_entry.options.get(
+                    CONF_DETECT_METERS, DEFAULT_DETECT_METERS
+                ),
+                CONF_DETECT_BATTERIES: self.config_entry.options.get(
+                    CONF_DETECT_BATTERIES, DEFAULT_DETECT_BATTERIES
+                ),
             }
 
         return self.async_show_form(
-            step_id = "init",
-            data_schema = vol.Schema(
+            step_id="init",
+            data_schema=vol.Schema(
                 {
                     vol.Optional(
-                        CONF_SCAN_INTERVAL, default=user_input[CONF_SCAN_INTERVAL]
+                        CONF_SCAN_INTERVAL,
+                        default=user_input[CONF_SCAN_INTERVAL],
                     ): vol.Coerce(int),
                     vol.Optional(
-                        CONF_SINGLE_DEVICE_ENTITY, default=user_input[CONF_SINGLE_DEVICE_ENTITY]
+                        CONF_SINGLE_DEVICE_ENTITY,
+                        default=user_input[CONF_SINGLE_DEVICE_ENTITY],
                     ): cv.boolean,
                     vol.Optional(
-                        CONF_KEEP_MODBUS_OPEN, default=user_input[CONF_KEEP_MODBUS_OPEN]
+                        CONF_KEEP_MODBUS_OPEN,
+                        default=user_input[CONF_KEEP_MODBUS_OPEN],
                     ): cv.boolean,
                     vol.Optional(
-                        CONF_DETECT_METERS, default=user_input[CONF_DETECT_METERS]
+                        CONF_DETECT_METERS,
+                        default=user_input[CONF_DETECT_METERS],
                     ): cv.boolean,
                     vol.Optional(
-                        CONF_DETECT_BATTERIES, default=user_input[CONF_DETECT_BATTERIES]
+                        CONF_DETECT_BATTERIES,
+                        default=user_input[CONF_DETECT_BATTERIES],
                     ): cv.boolean,
                 },
             ),
-            errors = errors
+            errors=errors,
         )

custom_components/solaredge_modbus_multi/const.py

@@ -28,7 +28,29 @@
 SUNSPEC_ACCUM_LIMIT = 4294967295
 SUNSPEC_NOT_IMPL_FLOAT32 = 0x7FC00000
 
-SUNSPEC_SF_RANGE = [-10, -9, -8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+SUNSPEC_SF_RANGE = [
+    -10,
+    -9,
+    -8,
+    -7,
+    -6,
+    -5,
+    -4,
+    -3,
+    -2,
+    -1,
+    0,
+    1,
+    2,
+    3,
+    4,
+    5,
+    6,
+    7,
+    8,
+    9,
+    10,
+]
 
 # parameter names per sunspec
 DEVICE_STATUS_DESC = {