fix: serial entries on one adapter now share a connection (#384)

The shared connection hub exists to serialize transactions and 'prevent RS485
cross-talk', and was created only for TCP entries. That had it backwards: an
RS485 bus is precisely where two uncoordinated masters collide.

Each serial entry opened its own client on the same adapter and paced itself
with a per-instance min_read_interval, which says nothing about what the other
entry is doing. Two inverters on one USB-RS485 adapter - the normal way to wire
a parallel SPF stack - interleaved frames on one physical bus with nothing
serializing them, giving random single-sample read failures on both units.

The hub is now transport-agnostic, keyed on device path for serial and
host:port for TCP. Serial gets its own buffer drain via pyserial's
reset_input_buffer(), since the TCP path's socket recv() loop raises on a
Serial object and would silently skip the flush.

Parity/stopbits/bytesize default to N/1/8, matching what the non-shared path
has always hardcoded, so moving to a shared connection cannot change framing.

Note the runtime name for the serial class is ModbusClient - ModbusSerialClient
is imported only under TYPE_CHECKING. The first cut of this used the latter and
would have raised NameError on the first connection; the whole existing suite
passed because nothing ever asked a serial hub to connect. The new tests
exercise that path.

Verified red without the fix.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: 0xAHA

RELEASENOTES.md

@@ -4,6 +4,20 @@
 
 ---
 
+## v1.7.0 (pre-release)
+
+> **Pre-release for testing.** v1.6.2 remains the stable release.
+
+- **Multiple inverters on one USB-RS485 adapter now share a single connection.** Until now
+  each entry opened its own serial client on the same adapter and paced only itself, so two
+  pollers interleaved their frames on one bus with nothing coordinating them — which shows up
+  as random, unexplained read failures on both inverters. Serial entries on the same device
+  path are now serialised behind one lock, the same way TCP entries on the same host:port
+  already were. **Only affects setups with two or more entries on one adapter**; single-entry
+  setups are unchanged, and TCP is untouched.
+
+---
+
 ## v1.6.9 (pre-release)
 
 Issues: #384

custom_components/growatt_modbus/__init__.py

@@ -17,6 +17,9 @@
     CONF_INVERTER_SERIES,
     CONF_REGISTER_MAP,
     CONF_CONNECTION_TYPE,
+    CONF_DEVICE_PATH,
+    CONF_BAUDRATE,
+    DEFAULT_BAUDRATE,
     CURRENT_DEVICE_STRUCTURE_VERSION,
     REGISTER_MAPS,
     WRITABLE_REGISTERS,
@@ -242,27 +245,48 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         _LOGGER.info("Removing stale number entity %s (WIT export_limit_w reg 203 not writable)", _stale_export_eid)
         entity_registry.async_remove(_stale_export_eid)
 
-    # Shared connection hub: all TCP entries on the same host:port share one ModbusTcpClient
-    # and a threading.Lock to serialize reads/writes and prevent RS485 cross-talk on the gateway.
+    # Shared connection hub: every entry on the same transport — the same host:port for TCP,
+    # the same device path for serial — shares one client and a threading.Lock that
+    # serializes reads and writes, preventing RS485 cross-talk.
+    #
+    # Serial was excluded from this until v1.7.0, which had it backwards: an RS485 bus is
+    # precisely where two uncoordinated masters collide. Each entry opened its own
+    # ModbusSerialClient on the same adapter and paced itself with a per-instance
+    # min_read_interval, which says nothing about what the other entry is doing. Two
+    # inverters on one USB-RS485 adapter is the normal way to wire a parallel SPF stack.
+    #
     # This is transparent for single-entry setups (hub refcount=1, no actual sharing).
     hub: SharedModbusConnection | None = None
     connection_type = entry.data.get(CONF_CONNECTION_TYPE, "tcp")
+    timeout = entry.options.get("timeout", 10)
+    connections = hass.data[DOMAIN].setdefault("_connections", {})
+
     if connection_type == "tcp":
         from homeassistant.const import CONF_HOST, CONF_PORT
         host = entry.data.get(CONF_HOST, "")
         port = entry.data.get(CONF_PORT, 502)
-        timeout = entry.options.get("timeout", 10)
-        hub_key = f"{host}:{port}"
-        connections = hass.data[DOMAIN].setdefault("_connections", {})
+        hub_key = f"{host}:{port}" if host else ""
+        hub_factory = lambda: SharedModbusConnection(host=host, port=port, timeout=timeout)
+    else:
+        device = entry.data.get(CONF_DEVICE_PATH, "")
+        baudrate = entry.data.get(CONF_BAUDRATE, DEFAULT_BAUDRATE)
+        # Keyed on the device path alone: baud rate is a property of the bus, so two entries
+        # disagreeing about it are misconfigured rather than entitled to separate clients.
+        hub_key = f"serial:{device}" if device else ""
+        hub_factory = lambda: SharedModbusConnection(
+            device=device, baudrate=baudrate, timeout=timeout
+        )
+
+    if hub_key:
         if hub_key not in connections:
-            connections[hub_key] = SharedModbusConnection(host=host, port=port, timeout=timeout)
+            connections[hub_key] = hub_factory()
             _LOGGER.debug("Created shared Modbus connection hub for %s", hub_key)
         hub = connections[hub_key]
         hub.acquire_ref()
         if hub._refcount > 1:
             _LOGGER.info(
                 "Shared Modbus connection mode: entry %s joined hub for %s (refcount=%d) — "
-                "RS485 gateway cross-talk prevention active",
+                "RS485 cross-talk prevention active",
                 entry.entry_id, hub_key, hub._refcount,
             )
 

custom_components/growatt_modbus/growatt_modbus.py

@@ -432,18 +432,43 @@ class GrowattData:
     serial_number: str = ""
 
 class SharedModbusConnection:
-    """Single ModbusTcpClient shared across multiple GrowattModbus instances on the same host:port.
+    """One Modbus client shared across every GrowattModbus instance on the same transport.
+
+    For TCP that means the same host:port; for serial, the same device path.
 
     Serializes all Modbus transactions with a threading.Lock (because _fetch_data runs in
-    executor threads, not on the asyncio event loop).  Reference-counted so the TCP socket
+    executor threads, not on the asyncio event loop).  Reference-counted so the connection
     stays open as long as at least one coordinator needs it.
+
+    **Serial was excluded from this until v1.7.0**, which was backwards. The hub's stated
+    purpose is preventing RS485 cross-talk, and an RS485 bus is exactly where two
+    uncoordinated masters collide: each config entry opened its own ModbusSerialClient on
+    the same adapter and paced itself with a per-instance `min_read_interval`, which says
+    nothing about what the other entry is doing. Two inverters on one USB-RS485 adapter —
+    the normal way to wire a parallel SPF stack — interleaved their frames on one physical
+    bus with nothing serializing them.
     """
 
-    def __init__(self, host: str, port: int, timeout: int = 10) -> None:
+    def __init__(
+        self,
+        host: str = "",
+        port: int = 502,
+        timeout: int = 10,
+        device: str = "",
+        baudrate: int = 9600,
+        parity: str = "N",
+        stopbits: int = 1,
+        bytesize: int = 8,
+    ) -> None:
         self.host = host
         self.port = port
+        self.device = device
+        self.baudrate = baudrate
+        self.parity = parity
+        self.stopbits = stopbits
+        self.bytesize = bytesize
         self._timeout = timeout
-        self._client: Optional['ModbusTcpClient'] = None
+        self._client: Optional[Union['ModbusTcpClient', 'ModbusSerialClient']] = None
         # Reentrant so a caller can hold the bus across a sequence of writes while the
         # individual write methods still take it themselves (#331). A plain Lock would
         # deadlock the moment write_batch() wrapped anything. Reentrancy is per-thread,
@@ -467,6 +492,15 @@ def __init__(self, host: str, port: int, timeout: int = 10) -> None:
         self.good_reads = 0
         self.malformed_reads = 0
 
+    @property
+    def is_serial(self) -> bool:
+        return bool(self.device)
+
+    @property
+    def connection_id(self) -> str:
+        """Identifier used in log lines — the device path, or host:port."""
+        return self.device if self.is_serial else f"{self.host}:{self.port}"
+
     # ------------------------------------------------------------------
     # Reference counting
     # ------------------------------------------------------------------
@@ -498,12 +532,30 @@ def release_ref(self) -> None:
     def ensure_connected(self) -> bool:
         """Connect if not already open; flush stale bytes on a new connection."""
         if self._client is None:
-            try:
-                self._client = ModbusTcpClient(host=self.host, port=self.port, timeout=self._timeout)
-            except TypeError:
-                self._client = ModbusTcpClient(self.host, self.port)
-                if hasattr(self._client, 'timeout'):
-                    self._client.timeout = self._timeout
+            if self.is_serial:
+                if not SERIAL_AVAILABLE:
+                    logger.error(
+                        "[SharedConn %s] pyserial/pymodbus serial support is not installed",
+                        self.connection_id,
+                    )
+                    return False
+                # NB: the runtime name is ModbusClient — ModbusSerialClient is imported only
+                # under TYPE_CHECKING and does not exist when this executes.
+                self._client = ModbusClient(
+                    port=self.device,
+                    baudrate=self.baudrate,
+                    parity=self.parity,
+                    stopbits=self.stopbits,
+                    bytesize=self.bytesize,
+                    timeout=self._timeout,
+                )
+            else:
+                try:
+                    self._client = ModbusTcpClient(host=self.host, port=self.port, timeout=self._timeout)
+                except TypeError:
+                    self._client = ModbusTcpClient(self.host, self.port)
+                    if hasattr(self._client, 'timeout'):
+                        self._client.timeout = self._timeout
 
         try:
             if hasattr(self._client, 'is_socket_open') and self._client.is_socket_open():
@@ -535,13 +587,34 @@ def reset(self, reason: str = "") -> None:
         a wedged connection.
         """
         logger.warning(
-            "[SharedConn %s:%s] Resetting connection%s",
-            self.host, self.port, f": {reason}" if reason else "",
+            "[SharedConn %s] Resetting connection%s",
+            self.connection_id, f": {reason}" if reason else "",
         )
         self.disconnect()
 
     def _flush_receive_buffer(self) -> None:
-        """Drain stale Modbus responses left in the adapter's TCP buffer after reconnect."""
+        """Drain stale Modbus responses left in the adapter's buffer after reconnect."""
+        if self.is_serial:
+            # pyserial exposes its own drain; the socket recv() path below does not apply
+            # and would raise on a Serial object.
+            serial_port = getattr(self._client, 'socket', None)
+            if serial_port is None:
+                return
+            try:
+                waiting = getattr(serial_port, 'in_waiting', 0)
+                serial_port.reset_input_buffer()
+                if waiting:
+                    logger.debug(
+                        "[SharedConn %s] Flushed %d stale bytes from serial input buffer",
+                        self.connection_id, waiting,
+                    )
+            except Exception as exc:
+                logger.debug(
+                    "[SharedConn %s] Serial buffer flush failed (non-critical): %s",
+                    self.connection_id, exc,
+                )
+            return
+
         sock = getattr(self._client, 'socket', None)
         if sock is None:
             transport = getattr(self._client, 'transport', None)
@@ -565,11 +638,11 @@ def _flush_receive_buffer(self) -> None:
                 sock.settimeout(original_timeout)
             if discarded:
                 logger.debug(
-                    "[SharedConn %s:%s] Flushed %d stale bytes from receive buffer after reconnect",
-                    self.host, self.port, discarded,
+                    "[SharedConn %s] Flushed %d stale bytes from receive buffer after reconnect",
+                    self.connection_id, discarded,
                 )
         except Exception as exc:
-            logger.debug("[SharedConn %s:%s] Buffer flush failed (non-critical): %s", self.host, self.port, exc)
+            logger.debug("[SharedConn %s] Buffer flush failed (non-critical): %s", self.connection_id, exc)
 
     # ------------------------------------------------------------------
     # Register access (slave_id passed per call, not stored on hub)
@@ -603,9 +676,9 @@ def _validate_registers(self, resp, start: int, count: int) -> Optional[list]:
 
         if len(registers) != count:
             logger.warning(
-                "[SharedConn %s:%s] Short/misaligned read at %d: got %d of %d registers — "
+                "[SharedConn %s] Short/misaligned read at %d: got %d of %d registers — "
                 "discarding frame and flushing buffer",
-                self.host, self.port, start, len(registers), count,
+                self.connection_id, start, len(registers), count,
             )
             # A misaligned stream stays misaligned — which is why the corrupt values
             # repeat byte-for-byte rather than varying. Draining the buffer here gives the
@@ -647,15 +720,15 @@ def read_input_registers(self, start: int, count: int, slave_id: int) -> Optiona
             except Exception as exc:
                 if attempt == 0 and self._begin_recovery():
                     logger.debug(
-                        "[SharedConn %s:%s] read_input_registers(%d, %d) transport error "
+                        "[SharedConn %s] read_input_registers(%d, %d) transport error "
                         "(%s) — resetting and retrying once",
-                        self.host, self.port, start, count, exc,
+                        self.connection_id, start, count, exc,
                     )
                     self.reset("transport error during block read")
                     if self.ensure_connected():
                         continue
-                logger.debug("[SharedConn %s:%s] read_input_registers(%d, %d, slave=%d) error: %s",
-                             self.host, self.port, start, count, slave_id, exc)
+                logger.debug("[SharedConn %s] read_input_registers(%d, %d, slave=%d) error: %s",
+                             self.connection_id, start, count, slave_id, exc)
                 return None
 
             return self._validate_registers(resp, start, count)
@@ -677,15 +750,15 @@ def read_holding_registers(self, start: int, count: int, slave_id: int) -> Optio
             except Exception as exc:
                 if attempt == 0 and self._begin_recovery():
                     logger.debug(
-                        "[SharedConn %s:%s] read_holding_registers(%d, %d) transport error "
+                        "[SharedConn %s] read_holding_registers(%d, %d) transport error "
                         "(%s) — resetting and retrying once",
-                        self.host, self.port, start, count, exc,
+                        self.connection_id, start, count, exc,
                     )
                     self.reset("transport error during block read")
                     if self.ensure_connected():
                         continue
-                logger.debug("[SharedConn %s:%s] read_holding_registers(%d, %d, slave=%d) error: %s",
-                             self.host, self.port, start, count, slave_id, exc)
+                logger.debug("[SharedConn %s] read_holding_registers(%d, %d, slave=%d) error: %s",
+                             self.connection_id, start, count, slave_id, exc)
                 return None
 
             return self._validate_registers(resp, start, count)
@@ -730,15 +803,15 @@ def write_register(self, register: int, value: int, slave_id: int) -> bool:
             except Exception as exc:
                 if attempt == 0 and self._begin_recovery():
                     logger.debug(
-                        "[SharedConn %s:%s] write_register(%d, %d) transport error (%s) — "
+                        "[SharedConn %s] write_register(%d, %d) transport error (%s) — "
                         "resetting and retrying once",
-                        self.host, self.port, register, value, exc,
+                        self.connection_id, register, value, exc,
                     )
                     self.reset("transport error during write")
                     if self.ensure_connected():
                         continue
-                logger.debug("[SharedConn %s:%s] write_register(%d, %d, slave=%d) error: %s",
-                             self.host, self.port, register, value, slave_id, exc)
+                logger.debug("[SharedConn %s] write_register(%d, %d, slave=%d) error: %s",
+                             self.connection_id, register, value, slave_id, exc)
                 self.disconnect()
                 return False
 
@@ -760,15 +833,15 @@ def write_registers(self, register: int, values: list, slave_id: int) -> bool:
             except Exception as exc:
                 if attempt == 0 and self._begin_recovery():
                     logger.debug(
-                        "[SharedConn %s:%s] write_registers(%d, %d values) transport error "
+                        "[SharedConn %s] write_registers(%d, %d values) transport error "
                         "(%s) — resetting and retrying once",
-                        self.host, self.port, register, len(values), exc,
+                        self.connection_id, register, len(values), exc,
                     )
                     self.reset("transport error during write")
                     if self.ensure_connected():
                         continue
-                logger.debug("[SharedConn %s:%s] write_registers(%d, slave=%d) error: %s",
-                             self.host, self.port, register, slave_id, exc)
+                logger.debug("[SharedConn %s] write_registers(%d, slave=%d) error: %s",
+                             self.connection_id, register, slave_id, exc)
                 self.disconnect()
                 return False
 

custom_components/growatt_modbus/manifest.json

@@ -14,5 +14,5 @@
     "pymodbus>=3.0.0",
     "pyserial>=3.4"
   ],
-  "version": "1.6.9"
+  "version": "1.7.0"
 }

docs/troubleshooting/rs485-gateways.md

@@ -157,6 +157,23 @@ The integration holds one socket per host:port across polls. It was suspected of
 
 ---
 
+## Two or more inverters on one adapter
+
+If you run several inverters as separate integration entries over **one** USB-RS485 adapter or one gateway, they share a single physical bus. Only one master can be talking at a time.
+
+The integration handles this for you: every entry pointing at the same device path (or the same host:port) shares one connection and one lock, so their reads are queued rather than interleaved. Nothing to configure.
+
+**Serial setups need v1.7.0 or later for this.** Before that, only TCP entries were coordinated — each serial entry opened its own client on the same adapter and paced only itself, which produced random read failures on *all* the inverters sharing that bus. If you have two entries on one adapter and see unexplained dropouts, this is the first thing to rule out.
+
+Two consequences worth knowing:
+
+- **Polls queue, they do not overlap.** With several inverters the effective cycle is the sum of their polls. If a single poll takes 8 s, three inverters need an interval comfortably above 24 s.
+- **A slow or failing inverter slows the others**, because they wait for the bus. If one entry is much worse than the rest, disable it and see whether the others recover — that isolates the problem device quickly.
+
+Raising the scan interval is the main lever here, exactly as it is for a single unit.
+
+---
+
 ## Gaps in the graph are the fix, not the fault
 
 Since **v1.6.6** a register that could not be read is published as **unavailable** for that poll, not as `0`.