Merge pull request #1230 from rany2/improve_lan_mode

Improve Local Mode

custom_components/sonoff/core/ewelink/__init__.py

@@ -101,7 +101,7 @@ async def send(
           ignored if params empty
         :param timeout_lan: optional custom LAN timeout
         """
-        seq = self.sequence()
+        seq = await self.sequence()
 
         if "parent" in device:
             main_device = device["parent"]

custom_components/sonoff/core/ewelink/base.py

@@ -35,20 +35,22 @@ class XDevice(TypedDict, total=False):
 class XRegistryBase:
     dispatcher: dict[str, list[Callable]] = None
     _sequence: int = 0
+    _sequence_lock: asyncio.Lock = asyncio.Lock()
 
     def __init__(self, session: ClientSession):
         self.dispatcher = {}
         self.session = session
 
     @staticmethod
-    def sequence() -> str:
+    async def sequence() -> str:
         """Return sequnce counter in ms. Always unique."""
-        t = int(time.time()) * 1000
-        if t > XRegistryBase._sequence:
-            XRegistryBase._sequence = t
-        else:
-            XRegistryBase._sequence += 1
-        return str(XRegistryBase._sequence)
+        t = time.time_ns() // 1_000_000
+        async with XRegistryBase._sequence_lock:
+            if t > XRegistryBase._sequence:
+                XRegistryBase._sequence = t
+            else:
+                XRegistryBase._sequence += 1
+            return str(XRegistryBase._sequence)
 
     def dispatcher_connect(self, signal: str, target: Callable) -> Callable:
         targets = self.dispatcher.setdefault(signal, [])

custom_components/sonoff/core/ewelink/cloud.py

@@ -248,7 +248,7 @@ async def send(
         self.last_ts = time.time()
 
         if sequence is None:
-            sequence = self.sequence()
+            sequence = await self.sequence()
         log += sequence
 
         # https://coolkit-technologies.github.io/eWeLink-API/#/en/APICenterV2?id=websocket-update-device-status

custom_components/sonoff/core/ewelink/local.py

@@ -5,6 +5,7 @@
 """
 import asyncio
 import base64
+import errno
 import ipaddress
 import json
 import logging
@@ -162,6 +163,7 @@ async def send(
         command: str = None,
         sequence: str = None,
         timeout: int = 5,
+        cre_retry_counter: int = 10,
     ):
         # known commands for DIY: switch, startup, pulse, sledonline
         # other commands: switch, switches, transmit, dimmable, light, fan
@@ -172,7 +174,7 @@ async def send(
             command = next(iter(params))
 
         payload = {
-            "sequence": sequence or self.sequence(),
+            "sequence": sequence or await self.sequence(),
             "deviceid": device["deviceid"],
             "selfApikey": "123",
             "data": params or {},
@@ -231,8 +233,31 @@ async def send(
             _LOGGER.debug(f"{log} !! Can't connect: {e}")
             return "E#CON"
 
+        except aiohttp.ClientOSError as e:
+            if e.errno != errno.ECONNRESET:
+                _LOGGER.debug(log, exc_info=e)
+                return "E#COE"  # ClientOSError
+
+            # This happens because the device's web server is not multi-threaded
+            # and can only process one request at a time. Therefore, if the
+            # device is busy processing another request, it will close the
+            # connection for the new request and we will get this error.
+            #
+            # It appears that the device takes some time to process a new request
+            # after the previous one was closed, which caused a locking approach
+            # to not work across different devices. Simply retrying on this error
+            # a few times seems to fortunately work reliably, so we'll do that.
+
+            _LOGGER.debug(f"{log} !! ConnectionResetError")
+            if cre_retry_counter > 0:
+                await asyncio.sleep(0.1)
+                return await self.send(
+                    device, params, command, sequence, timeout, cre_retry_counter - 1
+                )
+
+            return "E#CRE"  # ConnectionResetError
+
         except (
-            aiohttp.ClientOSError,
             aiohttp.ServerDisconnectedError,
             asyncio.CancelledError,
         ) as e: