Request/Response: change execute to be async method (#2142)

Author: ilkka-ollakka

pymodbus/pdu/bit_read_message.py

@@ -155,7 +155,7 @@ def __init__(self, address=None, count=None, slave=0, **kwargs):
         """
         ReadBitsRequestBase.__init__(self, address, count, slave, **kwargs)
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a read coils request against a datastore.
 
         Before running the request, we make sure that the request is in
@@ -223,7 +223,7 @@ def __init__(self, address=None, count=None, slave=0, **kwargs):
         """
         ReadBitsRequestBase.__init__(self, address, count, slave, **kwargs)
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a read discrete input request against a datastore.
 
         Before running the request, we make sure that the request is in

pymodbus/pdu/bit_write_message.py

@@ -79,7 +79,7 @@ def decode(self, data):
         self.address, value = struct.unpack(">HH", data)
         self.value = value == ModbusStatus.ON
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a write coil request against a datastore.
 
         :param context: The datastore to request from
@@ -213,7 +213,7 @@ def decode(self, data):
         values = unpack_bitstring(data[5:])
         self.values = values[:count]
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a write coils request against a datastore.
 
         :param context: The datastore to request from

pymodbus/pdu/diag_message.py

@@ -199,7 +199,7 @@ def __init__(self, data=0x0000, **kwargs):
         DiagnosticStatusRequest.__init__(self, **kwargs)
         self.message = data
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Raise if not implemented."""
         raise NotImplementedException("Diagnostic Message Has No Execute Method")
 
@@ -245,7 +245,7 @@ def __init__(self, message=b"\x00\x00", slave=None, **kwargs):
             raise ModbusException(f"message({type(message)}) must be bytes")
         self.message = message
 
-    def execute(self, *_args):
+    async def execute(self, *_args):
         """Execute the loopback request (builds the response).
 
         :returns: The populated loopback response message
@@ -301,7 +301,7 @@ def __init__(self, toggle=False, slave=None, **kwargs):
         else:
             self.message = [ModbusStatus.OFF]
 
-    def execute(self, *_args):
+    async def execute(self, *_args):
         """Clear event log and restart.
 
         :returns: The initialized response message
@@ -343,7 +343,7 @@ class ReturnDiagnosticRegisterRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0002
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -377,7 +377,7 @@ class ChangeAsciiInputDelimiterRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0003
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -412,7 +412,7 @@ class ForceListenOnlyModeRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0004
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -451,7 +451,7 @@ class ClearCountersRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x000A
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -482,7 +482,7 @@ class ReturnBusMessageCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x000B
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -515,7 +515,7 @@ class ReturnBusCommunicationErrorCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x000C
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -548,7 +548,7 @@ class ReturnBusExceptionErrorCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x000D
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -581,7 +581,7 @@ class ReturnSlaveMessageCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x000E
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -614,7 +614,7 @@ class ReturnSlaveNoResponseCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x000F
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -648,7 +648,7 @@ class ReturnSlaveNAKCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0010
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -682,7 +682,7 @@ class ReturnSlaveBusyCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0011
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -717,7 +717,7 @@ class ReturnSlaveBusCharacterOverrunCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0012
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -750,7 +750,7 @@ class ReturnIopOverrunCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0013
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -783,7 +783,7 @@ class ClearOverrunCountRequest(DiagnosticStatusSimpleRequest):
 
     sub_function_code = 0x0014
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message
@@ -834,7 +834,7 @@ def get_response_pdu_size(self):
             data = 0
         return 1 + 2 + 2 + 2 + data
 
-    def execute(self, *args):
+    async def execute(self, *args):
         """Execute the diagnostic request on the given device.
 
         :returns: The initialized response message

pymodbus/pdu/mei_message.py

@@ -84,7 +84,7 @@ def decode(self, data):
         params = struct.unpack(">BBB", data)
         self.sub_function_code, self.read_code, self.object_id = params
 
-    def execute(self, _context):
+    async def execute(self, _context):
         """Run a read exception status request against the store.
 
         :returns: The populated response

pymodbus/pdu/other_message.py

@@ -54,7 +54,7 @@ def decode(self, data):
         :param data: The incoming data
         """
 
-    def execute(self, _context=None):
+    async def execute(self, _context=None):
         """Run a read exception status request against the store.
 
         :returns: The populated response
@@ -159,7 +159,7 @@ def decode(self, data):
         :param data: The incoming data
         """
 
-    def execute(self, _context=None):
+    async def execute(self, _context=None):
         """Run a read exception status request against the store.
 
         :returns: The populated response
@@ -270,7 +270,7 @@ def decode(self, data):
         :param data: The incoming data
         """
 
-    def execute(self, _context=None):
+    async def execute(self, _context=None):
         """Run a read exception status request against the store.
 
         :returns: The populated response
@@ -395,7 +395,7 @@ def decode(self, data):
         :param data: The incoming data
         """
 
-    def execute(self, context=None):
+    async def execute(self, context=None):
         """Run a report slave id request against the store.
 
         :returns: The populated response

pymodbus/pdu/pdu.py

@@ -252,7 +252,7 @@ def decode(self, _data):
     def encode(self):
         """Decode so this failure will run correctly."""
 
-    def execute(self, _context):
+    async def execute(self, _context):
         """Build an illegal function request error response.
 
         :returns: The error response packet

pymodbus/pdu/register_read_message.py

@@ -139,7 +139,7 @@ def __init__(self, address=None, count=None, slave=0, **kwargs):
         """
         super().__init__(address, count, slave, **kwargs)
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a read holding request against a datastore.
 
         :param context: The datastore to request from
@@ -200,7 +200,7 @@ def __init__(self, address=None, count=None, slave=0, **kwargs):
         """
         super().__init__(address, count, slave, **kwargs)
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a read input request against a datastore.
 
         :param context: The datastore to request from
@@ -310,7 +310,7 @@ def decode(self, data):
             register = struct.unpack(">H", data[i : i + 2])[0]
             self.write_registers.append(register)
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a write single register request against a datastore.
 
         :param context: The datastore to request from

pymodbus/pdu/register_write_message.py

@@ -57,7 +57,7 @@ def decode(self, data):
         """
         self.address, self.value = struct.unpack(">HH", data)
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a write single register request against a datastore.
 
         :param context: The datastore to request from
@@ -200,7 +200,7 @@ def decode(self, data):
         for idx in range(5, (self.count * 2) + 5, 2):
             self.values.append(struct.unpack(">H", data[idx : idx + 2])[0])
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a write single register request against a datastore.
 
         :param context: The datastore to request from
@@ -321,7 +321,7 @@ def decode(self, data):
         """
         self.address, self.and_mask, self.or_mask = struct.unpack(">HHH", data)
 
-    def execute(self, context):
+    async def execute(self, context):
         """Run a mask write register request against the store.
 
         :param context: The datastore to request from

pymodbus/server/async_io.py

@@ -184,17 +184,10 @@ async def _async_execute(self, request, *addr):
                 # if broadcasting then execute on all slave contexts,
                 # note response will be ignored
                 for slave_id in self.server.context.slaves():
-                    response = request.execute(self.server.context[slave_id])
-                    # Temporary check while we move execute to async method
-                    if asyncio.iscoroutine(response):
-                        response = await response
+                    response = await request.execute(self.server.context[slave_id])
             else:
                 context = self.server.context[request.slave_id]
-                response = request.execute(context)
-
-                # Temporary check while we move execute to async method
-                if asyncio.iscoroutine(response):
-                    response = await response
+                response = await request.execute(context)
 
         except NoSuchSlaveException:
             Log.error("requested slave does not exist: {}", request.slave_id)

test/sub_client/test_client.py

@@ -392,7 +392,7 @@ def __init__(self, base, req, retries=0):
     async def delayed_resp(self):
         """Send a response to a received packet."""
         await asyncio.sleep(0.05)
-        resp = self.req.execute(self.ctx)
+        resp = await self.req.execute(self.ctx)
         pkt = self.base.ctx.framer.buildPacket(resp)
         self.base.ctx.data_received(pkt)