修复异步和循环问题

Author: WalterDSU

custom_components/peacefair_energy/__init__.py

@@ -3,6 +3,7 @@
 import voluptuous as vol
 import os
 from datetime import timedelta
+import asyncio
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 from homeassistant.core import HomeAssistant
 from .const import(
@@ -72,7 +73,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry):
     await hass.config_entries.async_forward_entry_setups(config_entry, ["sensor"])
     hass.data[config_entry.entry_id][UN_SUBDISCRIPT] = config_entry.add_update_listener(update_listener)
 
-    def service_handle(service):
+    async def async_service_handle(service):
         entity_id = service.data[ATTR_ENTITY_ID]
         energy_sensor = next(
             (sensor for sensor in hass.data[DOMAIN][ENERGY_SENSOR] if sensor.entity_id == entity_id),
@@ -85,20 +86,19 @@ def service_handle(service):
             coordinator = hass.data[config_entry.entry_id][COORDINATOR]
             if coordinator is not None:
                 coordinator.reset_energy()
-                energy_sensor.reset()
+                await energy_sensor.reset()  # 等待异步方法完成
 
     hass.services.async_register(
         DOMAIN,
         SERVICE_RESET_ENERGY,
-        service_handle,
+        async_service_handle,
         schema=RESET_ENERGY_SCHEMA,
     )
 
     return True
 
 
 async def async_unload_entry(hass: HomeAssistant, config_entry):
-
     await hass.config_entries.async_forward_entry_unloads(config_entry, ["sensor"])
 
     host = config_entry.data[CONF_HOST]
@@ -113,15 +113,22 @@ async def async_unload_entry(hass: HomeAssistant, config_entry):
     if unsub is not None:
         unsub()
     hass.data.pop(config_entry.entry_id)
+    
+    # 异步执行文件操作
     storage_path = hass.config.path(f"{STORAGE_PATH}")
     record_file = hass.config.path(f"{STORAGE_PATH}/{config_entry.entry_id}_state.json")
     reset_file = hass.config.path(f"{STORAGE_PATH}/{DOMAIN}_reset.json")
-    if os.path.exists(record_file):
-        os.remove(record_file)
-    if os.path.exists(reset_file):
-        os.remove(reset_file)
-    if len(os.listdir(storage_path)) == 0:
-        os.rmdir(storage_path)
+    
+    def remove_files():
+        if os.path.exists(record_file):
+            os.remove(record_file)
+        if os.path.exists(reset_file):
+            os.remove(reset_file)
+        if len(os.listdir(storage_path)) == 0:
+            os.rmdir(storage_path)
+    
+    await hass.async_add_executor_job(remove_files)
+    
     return True
 
 
@@ -156,5 +163,9 @@ async def _async_update_data(self):
             data = data_update
             _LOGGER.debug(f"Got Data {data}")
             if self._updates is not None:
-                self._updates()
+                # 确保异步方法被正确调用
+                if asyncio.iscoroutinefunction(self._updates):
+                    await self._updates()
+                else:
+                    self._updates()
         return data

custom_components/peacefair_energy/const.py

@@ -8,7 +8,7 @@
 UN_SUBDISCRIPT = "un_subdiscript"
 DEVICE_CLASS_FREQUENCY = "frequency"
 DEVICES = "devices"
-VERSION = "0.7.0"
+VERSION = "3.11.1"
 GATHER_TIME = "gather_time"
 PROTOCOLS = {
     "ModbusRTU Over UDP/IP": "rtuoverudp",

custom_components/peacefair_energy/manifest.json

@@ -1,12 +1,12 @@
 {
   "domain": "peacefair_energy",
   "name": "Peacefair Energy Monitor",
-  "version": "v0.7.5",
+  "version": "v3.11.1",
   "config_flow": true,
   "documentation": "https://github.com/georgezhao2010/peacefair_energy",
   "issue_tracker": "https://github.com/georgezhao2010/peacefair_energy/issue",
   "iot_class": "local_polling",
   "dependencies": [],
   "codeowners": ["@georgezhao2010"],
-  "requirements": ["pymodbus==3.6.9"]
+  "requirements": ["pymodbus"]
 }

custom_components/peacefair_energy/modbus.py

@@ -1,8 +1,10 @@
 import logging
 import asyncio
 from pymodbus.client import ModbusTcpClient, ModbusUdpClient
-from pymodbus.transaction import ModbusRtuFramer, ModbusIOException
-from pymodbus.pdu import ModbusRequest
+from pymodbus.framer.rtu import FramerRTU
+from pymodbus.transaction.transaction import ModbusIOException
+from pymodbus.pdu import ModbusPDU
+#from pymodbus.exceptions import ModbusException
 import threading
 
 try:
@@ -35,6 +37,29 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+class ModbusRequest(ModbusPDU):
+    """Base class for a modbus request PDU."""
+
+    function_code = -1
+
+    def __init__(self, slave, transaction, skip_encode):
+        """Proxy to the lower level initializer.
+
+        :param slave: Modbus slave slave ID
+        """
+        super().__init__(slave, transaction, skip_encode)
+        self.fut = None
+
+    def doException(self, exception):
+        """Build an error response based on the function.
+
+        :param exception: The exception to return
+        :raises: An exception response
+        """
+        exc = ExceptionResponse(self.function_code, exception)
+        Log.error("Exception response {}", exc)
+        return exc
+
 class ModbusResetEnergyRequest(ModbusRequest):
     _rtu_frame_size = 4
     function_code = 0x42
@@ -58,19 +83,19 @@ def __init__(self, protocol, host, port, slave):
             self._client = ModbusTcpClient(
                 host=host,
                 port=port,
-                framer=ModbusRtuFramer,
+                framer='rtu',
                 timeout=2,
-                retry_on_empty=True,
-                retry_on_invalid=False
+                #retry_on_empty=True,
+                #retry_on_invalid=False
             )
         elif protocol == "rtuoverudp":
             self._client = ModbusUdpClient(
                 host=host,
                 port=port,
-                framer=ModbusRtuFramer,
+                framer='rtu',
                 timeout=2,
-                retry_on_empty=False,
-                retry_on_invalid=False
+                #retry_on_empty=False,
+                #retry_on_invalid=False
             )
 
     def connect(self):
@@ -81,27 +106,27 @@ def close(self):
         with self._lock:
             self._client.close()
 
-    # 新增同步版本，供线程池调用
-    def read_input_registers_sync(self, address, count):
-        with self._lock:
-            kwargs = {"slave": self._slave}
-            return self._client.read_input_registers(address, count, **kwargs)
-
-    # 异步版本，交给线程池执行，避免阻塞事件循环
-    async def read_input_registers(self, address, count):
-        loop = asyncio.get_running_loop()
-        return await loop.run_in_executor(
-            None,
-            self.read_input_registers_sync,
-            address,
-            count
-        )
-
-    def reset_energy(self):
-        with self._lock:
-            kwargs = {"slave": self._slave}
-            request = ModbusResetEnergyRequest(**kwargs)
-            self._client.execute(request)
+    # 新增同步版本，供线程池调用
+    def read_input_registers_sync(self, address, count):
+        with self._lock:
+            kwargs = {"device_id": self._slave}
+            return self._client.read_input_registers(address=address, count=count, **kwargs)
+
+    # 异步版本，交给线程池执行，避免阻塞事件循环
+    async def read_input_registers(self, address, count):
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(
+            None,
+            self.read_input_registers_sync,
+            address,
+            count
+        )
+
+    def reset_energy(self):
+        with self._lock:
+            kwargs = {"slave": self._slave}
+            request = ModbusResetEnergyRequest(**kwargs)
+            self._client.execute(request)
 
     # 异步版本info_gather
     async def info_gather(self):