Enhance configuration, add Bluetooth status and Venus D support (#140)

Author: ViperRNMC

custom_components/marstek_modbus/config_flow.py

@@ -98,10 +98,11 @@ async def async_step_user(self, user_input=None):
             except (socket.gaierror, TypeError):
                 errors["base"] = "invalid_host"
             else:
-                # Prevent duplicate entries for same host and unit_id
+                # Prevent duplicate entries for same host, port and unit_id
                 for entry in self._async_current_entries():
                     if (
                         entry.data.get(CONF_HOST) == host
+                        and entry.data.get(CONF_PORT) == port
                         and entry.data.get(CONF_UNIT_ID) == unit_id
                     ):
                         return self.async_abort(reason="already_configured")

custom_components/marstek_modbus/const.py

@@ -22,7 +22,8 @@
 SUPPORTED_VERSIONS = [
     "E v1/v2", 
     "E v3",
-    "D"]
+    "D",
+    "A"]
 
 # Note: register loading logic (get_registers_for_version) was moved to
 # `register_loader.py` to keep `const.py` focused on constants only.

custom_components/marstek_modbus/coordinator.py

@@ -14,6 +14,7 @@
 from .const import DEFAULT_SCAN_INTERVALS, SUPPORTED_VERSIONS, DEFAULT_UNIT_ID
 
 from .helpers.modbus_client import MarstekModbusClient
+from pathlib import Path
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -110,6 +111,10 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry):
         self._connection_suspended = False
         self._suspension_reset_time = None
 
+        self._consecutive_timeout_cycles = 0
+        self._max_consecutive_timeout_cycles = 3
+        self._timeout_ratio_reconnect_threshold = 0.5
+        
         # Connection health tracking for diagnostics
         self._last_successful_read = None
         self._connection_established_at = None
@@ -215,8 +220,14 @@ async def async_init(self):
             _LOGGER.info("Successfully connected to Modbus device at %s:%d", self.host, self.port)
         return connected
 
-    async def async_read_value(self, sensor: dict, key: str):
-        """Helper to read a single sensor value from Modbus with logging and type checking."""
+    async def async_read_value(self, sensor: dict, key: str, track_failure: bool = True):
+        """Helper to read a single sensor value from Modbus with logging and type checking.
+
+        Args:
+            sensor: sensor definition dict
+            key: the sensor key
+            track_failure: if False, timeouts will not count towards timeout metrics
+        """
         entity_type = self._entity_types.get(key, get_entity_type(sensor))
 
          # Determine scale and unit
@@ -260,9 +271,11 @@ async def async_read_value(self, sensor: dict, key: str):
                 return None
 
         except asyncio.TimeoutError:
+            if track_failure:
+                self._timeouts_in_cycle = getattr(self, "_timeouts_in_cycle", 0) + 1
             _LOGGER.warning(
-                "Timeout reading %s '%s' at register %d - connection may be slow or incorrect",
-                entity_type, key, sensor["register"]
+                "Timeout reading %s '%s' at register %d from %s:%d - connection may be slow or incorrect",
+                entity_type, key, sensor["register"], self.client.host, self.client.port
             )
             return None
         except Exception as e:
@@ -348,6 +361,7 @@ async def _async_update_data(self):
         # Track if we actually attempted any reads (not just skipped due to intervals)
         attempted_reads = 0
         successful_reads = 0
+        self._timeouts_in_cycle = 0
 
         # Connection throttling: if too many failures, temporarily stop attempting connections
         if self._connection_suspended:
@@ -358,14 +372,7 @@ async def _async_update_data(self):
 
                 # Force reconnect after suspension
                 try:
-                    _LOGGER.debug("Closing existing connection before reconnect")
-                    await self.client.async_close()
-                except Exception as exc:
-                    _LOGGER.debug("Error closing client during reconnect: %s", exc)
-                
-                try:
-                    _LOGGER.info("Attempting to reconnect to %s:%d", self.host, self.port)
-                    connected = await self.client.async_connect()
+                    connected = await self.client.async_reconnect()
                     if connected:
                         _LOGGER.info("Successfully reconnected after suspension")
                     else:
@@ -464,6 +471,7 @@ async def _async_update_data(self):
 
         # Connection retry logic: only track failures if we actually attempted reads
         if attempted_reads > 0:
+            timeout_reads = int(getattr(self, "_timeouts_in_cycle", 0) or 0)
             if successful_reads > 0:
                 # At least some data successfully retrieved - reset failure counter
                 if self._consecutive_failures > 0:
@@ -472,18 +480,46 @@ async def _async_update_data(self):
                 self._consecutive_failures = 0
                 self._connection_suspended = False
                 self._last_successful_read = now
+                
+                if timeout_reads and (timeout_reads / attempted_reads) >= self._timeout_ratio_reconnect_threshold:
+                    self._consecutive_timeout_cycles += 1
+                    _LOGGER.warning(
+                        "High timeout rate detected (%d/%d) - consecutive timeout cycles: %d/%d",
+                        timeout_reads,
+                        attempted_reads,
+                        self._consecutive_timeout_cycles,
+                        self._max_consecutive_timeout_cycles,
+                    )
+                else:
+                    self._consecutive_timeout_cycles = 0
+                
+                if self._consecutive_timeout_cycles >= self._max_consecutive_timeout_cycles:
+                    try:
+                        _LOGGER.info(
+                            "Attempting reconnect due to repeated timeouts (%d/%d cycles)",
+                            self._consecutive_timeout_cycles,
+                            self._max_consecutive_timeout_cycles,
+                        )
+                        connected = await self.client.async_reconnect()
+                        if connected:
+                            _LOGGER.info("Successfully reconnected after repeated timeouts")
+                            self._consecutive_timeout_cycles = 0
+                            self._connection_established_at = now
+                        else:
+                            _LOGGER.warning("Reconnect attempt after repeated timeouts failed")
+                    except Exception as exc:
+                        _LOGGER.error("Exception during reconnect after repeated timeouts: %s", exc)
             elif successful_reads == 0:
                 # We attempted reads but ALL failed - connection issue
                 self._consecutive_failures += 1
                 _LOGGER.warning("All read attempts failed (%d/%d) - consecutive failures: %d/%d",
                               successful_reads, attempted_reads, 
                               self._consecutive_failures, self._max_consecutive_failures)
 
-                # Try to reconnect immediately on failure
+                # Try to reconnect immediately on failure (use reconnect helper)
                 try:
                     _LOGGER.info("Attempting immediate reconnection after read failures")
-                    await self.client.async_close()
-                    connected = await self.client.async_connect()
+                    connected = await self.client.async_reconnect()
                     if connected:
                         _LOGGER.info("Successfully reconnected")
                         self._consecutive_failures = 0
@@ -502,6 +538,7 @@ async def _async_update_data(self):
                         "Will retry in 5 minutes to prevent resource exhaustion.",
                         self._consecutive_failures
                     )
+                self._consecutive_timeout_cycles = 0
         else:
             _LOGGER.debug("No sensors due for update in this cycle")
 
@@ -567,22 +604,90 @@ def get_registers(version: str):
             % (version_raw, ", ".join(sorted(allowed)))
         )
 
-    # Map the validated version token to the correct registers module.
-    # Support the new tokens 'e v1/v2' and 'e v3'.
+    def _normalize_section(section):
+        """Convert mapping-based sections into the legacy list-of-dicts format."""
+        if isinstance(section, dict):
+            normalized = []
+            for key, value in section.items():
+                entry = dict(value or {})
+                entry.setdefault("key", key)
+                normalized.append(entry)
+            return normalized
+        if isinstance(section, list):
+            return section
+        return []
+
+    # Prefer YAML-based register definitions placed in the `registers/` folder.
+    # Map version tokens to YAML filenames.
+    filename_map = {
+        "e v1/v2": "e_v12.yaml",
+        "e v3": "e_v3.yaml",
+        "d": "d.yaml",
+        "a": "a.yaml",
+    }
+
+    yaml_filename = filename_map.get(version)
+    if yaml_filename:
+        yaml_path = Path(__file__).parent / "registers" / yaml_filename
+        if yaml_path.exists():
+            try:
+                import yaml
+
+                with open(yaml_path, "r", encoding="utf-8") as fh:
+                    data = yaml.safe_load(fh) or {}
+
+                return {
+                    "SENSOR_DEFINITIONS": _normalize_section(data.get("SENSOR_DEFINITIONS")),
+                    "BINARY_SENSOR_DEFINITIONS": _normalize_section(data.get("BINARY_SENSOR_DEFINITIONS")),
+                    "SELECT_DEFINITIONS": _normalize_section(data.get("SELECT_DEFINITIONS")),
+                    "SWITCH_DEFINITIONS": _normalize_section(data.get("SWITCH_DEFINITIONS")),
+                    "NUMBER_DEFINITIONS": _normalize_section(data.get("NUMBER_DEFINITIONS")),
+                    "BUTTON_DEFINITIONS": _normalize_section(data.get("BUTTON_DEFINITIONS")),
+                    "EFFICIENCY_SENSOR_DEFINITIONS": _normalize_section(
+                        data.get("EFFICIENCY_SENSOR_DEFINITIONS")
+                    ),
+                    "STORED_ENERGY_SENSOR_DEFINITIONS": _normalize_section(
+                        data.get("STORED_ENERGY_SENSOR_DEFINITIONS")
+                    ),
+                }
+            except Exception as e:
+                _LOGGER.warning("Failed to load YAML registers %s: %s", yaml_path, e)
+
+    # Fall back to legacy Python modules if YAML not present or failed to load
     if version == "e v1/v2":
         from . import registers_v12 as registers
     elif version == "e v3":
         from . import registers_v3 as registers
     elif version == "d":
-        from . import registers_v12 as registers
-        
+        from . import registers_d as registers
+    elif version == "a":
+        # No legacy Python module for A exists; return empty definitions as fallback
+        registers = None
+
+    if registers:
+        return {
+            "SENSOR_DEFINITIONS": getattr(registers, "SENSOR_DEFINITIONS", []),
+            "BINARY_SENSOR_DEFINITIONS": getattr(registers, "BINARY_SENSOR_DEFINITIONS", []),
+            "SELECT_DEFINITIONS": getattr(registers, "SELECT_DEFINITIONS", []),
+            "SWITCH_DEFINITIONS": getattr(registers, "SWITCH_DEFINITIONS", []),
+            "NUMBER_DEFINITIONS": getattr(registers, "NUMBER_DEFINITIONS", []),
+            "BUTTON_DEFINITIONS": getattr(registers, "BUTTON_DEFINITIONS", []),
+            "EFFICIENCY_SENSOR_DEFINITIONS": getattr(
+                registers, "EFFICIENCY_SENSOR_DEFINITIONS", []
+            ),
+            "STORED_ENERGY_SENSOR_DEFINITIONS": getattr(
+                registers, "STORED_ENERGY_SENSOR_DEFINITIONS", []
+            ),
+        }
+
+    # Default empty return if nothing found
     return {
-        "SENSOR_DEFINITIONS": getattr(registers, "SENSOR_DEFINITIONS", []),
-        "BINARY_SENSOR_DEFINITIONS": getattr(registers, "BINARY_SENSOR_DEFINITIONS", []),
-        "SELECT_DEFINITIONS": getattr(registers, "SELECT_DEFINITIONS", []),
-        "SWITCH_DEFINITIONS": getattr(registers, "SWITCH_DEFINITIONS", []),
-        "NUMBER_DEFINITIONS": getattr(registers, "NUMBER_DEFINITIONS", []),
-        "BUTTON_DEFINITIONS": getattr(registers, "BUTTON_DEFINITIONS", []),
-        "EFFICIENCY_SENSOR_DEFINITIONS": getattr(registers, "EFFICIENCY_SENSOR_DEFINITIONS", []),
-        "STORED_ENERGY_SENSOR_DEFINITIONS": getattr(registers, "STORED_ENERGY_SENSOR_DEFINITIONS", []),
+        "SENSOR_DEFINITIONS": [],
+        "BINARY_SENSOR_DEFINITIONS": [],
+        "SELECT_DEFINITIONS": [],
+        "SWITCH_DEFINITIONS": [],
+        "NUMBER_DEFINITIONS": [],
+        "BUTTON_DEFINITIONS": [],
+        "EFFICIENCY_SENSOR_DEFINITIONS": [],
+        "STORED_ENERGY_SENSOR_DEFINITIONS": [],
     }

custom_components/marstek_modbus/helpers/modbus_client.py

@@ -150,6 +150,38 @@ async def async_close(self) -> None:
             except Exception:
                 pass
 
+    async def async_reconnect(self) -> bool:
+        """Reconnect to the Modbus TCP server by closing and re-opening the connection."""
+        async with self._request_lock:
+            _LOGGER.info("Reconnecting to Modbus server at %s:%s", self.host, self.port)
+
+            try:
+                try:
+                    await self.async_close()
+                except Exception as e:
+                    _LOGGER.debug("Error closing Modbus client during reconnect: %s", e)
+
+                try:
+                    connected = await self.async_connect()
+                except Exception as e:
+                    _LOGGER.warning(
+                        "Exception while reconnecting to Modbus server at %s:%s: %s",
+                        self.host,
+                        self.port,
+                        e,
+                    )
+                    return False
+
+                if connected:
+                    _LOGGER.info("Reconnected to Modbus server at %s:%s", self.host, self.port)
+                else:
+                    _LOGGER.warning("Reconnect failed to Modbus server at %s:%s", self.host, self.port)
+
+                return connected
+            except Exception as e:
+                _LOGGER.warning("Unhandled exception during reconnect: %s", e)
+                return False
+
     async def async_read_register(
         self,
         register: int,
@@ -265,14 +297,15 @@ async def async_read_register(
                 else:
                     regs = result.registers
                     _LOGGER.debug(
-                        "Requesting register %d (0x%04X) for sensor '%s' (type: %s, count: %s)",
+                        "Requesting register %d (0x%04X) from '%s' for sensor '%s' (type: %s, count: %s)",
                         register,
                         register,
+                        self.host,
                         sensor_key or 'unknown',
                         data_type,
                         count,
                     )
-                    _LOGGER.debug("Received data from register %d (0x%04X): %s", register, register, regs)
+                    _LOGGER.debug("Received data from '%s' for register %d (0x%04X): %s", self.host, register, register, regs)
 
                     if data_type == "int16":
                         val = regs[0]

custom_components/marstek_modbus/manifest.json

@@ -1,7 +1,7 @@
 {
   "domain": "marstek_modbus",
   "name": "Marstek Venus Modbus",
-  "version": "2026.1.1",
+  "version": "2026.3.1",
   "config_flow": true,
   "documentation": "https://github.com/viperrnmc/marstek_venus_modbus",
   "requirements": ["pymodbus>=3.9.2"],

custom_components/marstek_modbus/number.py

@@ -148,7 +148,7 @@ async def async_set_native_value(self, value: float) -> None:
         # Only refresh if write failed to get actual device state
         if not success:
             _LOGGER.debug("Write failed for %s, refreshing to get actual state", self._key)
-            await self.coordinator.async_read_value(self.definition, self._key)
+            await self.coordinator.async_read_value(self.definition, self._key, track_failure=False)
 
     @property
     def device_info(self) -> dict: