Merge pull request #327 from riptideio/#326-Use-Existing-Loop

#326 Handle asyncio loop

Author: dhoomakethu

README.rst

@@ -104,6 +104,15 @@ permissions or a virtualenv currently running)::
     easy_install -U pymodbus
     pip install  -U pymodbus
 
+To Install pymodbus with twisted support run
+.. code-block:: python
+    pip install -U pymodbus[twisted]
+
+To Install pymodbus with tornado support run
+
+.. code-block:: python
+    pip install -U pymodbus[tornado]
+
 Otherwise you can pull the trunk source and install from there::
 
     git clone git://github.com/bashwork/pymodbus.git

examples/common/async_asyncio_client.py

@@ -25,6 +25,8 @@
     sys.stderr("This example needs to be run only on python 3.4 and above")
     sys.exit(1)
 
+from threading import Thread
+import time
 # --------------------------------------------------------------------------- #
 # configure the client logging
 # --------------------------------------------------------------------------- #
@@ -129,8 +131,81 @@ async def start_async_test(client):
     await asyncio.sleep(1)
 
 
-if __name__ == '__main__':
+def run_with_not_running_loop():
+    """
+    A loop is created and is passed to ModbusClient factory to be used.
 
+    :return:
+    """
+    log.debug("Running Async client with asyncio loop not yet started")
+    log.debug("------------------------------------------------------")
+    loop = asyncio.new_event_loop()
+    assert not loop.is_running()
+    new_loop, client = ModbusClient(schedulers.ASYNC_IO, port=5020, loop=loop)
+    loop.run_until_complete(start_async_test(client.protocol))
+    loop.close()
+    log.debug("--------------RUN_WITH_NOT_RUNNING_LOOP---------------")
+    log.debug("")
+
+
+def run_with_already_running_loop():
+    """
+    An already running loop is passed to ModbusClient Factory
+    :return:
+    """
+    log.debug("Running Async client with asyncio loop already started")
+    log.debug("------------------------------------------------------")
+
+    def done(future):
+        log.info("Done !!!")
+
+    def start_loop(loop):
+        """
+        Start Loop
+        :param loop:
+        :return:
+        """
+        asyncio.set_event_loop(loop)
+        loop.run_forever()
+
+    loop = asyncio.new_event_loop()
+    t = Thread(target=start_loop, args=[loop])
+    t.daemon = True
+    # Start the loop
+    t.start()
+    assert loop.is_running()
+    asyncio.set_event_loop(loop)
+    loop, client = ModbusClient(schedulers.ASYNC_IO, port=5020, loop=loop)
+    future = asyncio.run_coroutine_threadsafe(
+        start_async_test(client.protocol), loop=loop)
+    future.add_done_callback(done)
+    while not future.done():
+        time.sleep(0.1)
+    loop.stop()
+    log.debug("--------DONE RUN_WITH_ALREADY_RUNNING_LOOP-------------")
+    log.debug("")
+
+
+def run_with_no_loop():
+    """
+    ModbusClient Factory creates a loop.
+    :return:
+    """
     loop, client = ModbusClient(schedulers.ASYNC_IO, port=5020)
     loop.run_until_complete(start_async_test(client.protocol))
     loop.close()
+
+
+if __name__ == '__main__':
+    # Run with No loop
+    log.debug("Running Async client")
+    log.debug("------------------------------------------------------")
+    run_with_no_loop()
+
+    # Run with loop not yet started
+    run_with_not_running_loop()
+
+    # Run with already running loop
+    run_with_already_running_loop()
+    log.debug("---------------------RUN_WITH_NO_LOOP-----------------")
+    log.debug("")

pymodbus/client/async/factory/tcp.py

@@ -91,10 +91,17 @@ def async_io_factory(host="127.0.0.1", port=Defaults.Port, framer=None,
     """
     import asyncio
     from pymodbus.client.async.asyncio import init_tcp_client
-    loop = kwargs.get("loop") or asyncio.get_event_loop()
+    loop = kwargs.get("loop") or asyncio.new_event_loop()
     proto_cls = kwargs.get("proto_cls", None)
-    cor = init_tcp_client(proto_cls, loop, host, port)
-    client = loop.run_until_complete(asyncio.gather(cor))[0]
+    if not loop.is_running():
+        asyncio.set_event_loop(loop)
+        cor = init_tcp_client(proto_cls, loop, host, port)
+        client = loop.run_until_complete(asyncio.gather(cor))[0]
+    else:
+        cor = init_tcp_client(proto_cls, loop, host, port)
+        future = asyncio.run_coroutine_threadsafe(cor, loop=loop)
+        client = future.result()
+
     return loop, client
 
 

pymodbus/payload.py

@@ -124,6 +124,17 @@ def to_registers(self):
         _logger.debug(payload)
         return payload
 
+    def to_coils(self):
+        """Convert the payload buffer into a coil
+        layout that can be used as a context block.
+
+        :returns: The coil layout to use as a block
+        """
+        payload = self.to_registers()
+        coils = [bool(int(bit)) for reg
+                 in payload[1:] for bit in format(reg, '016b')]
+        return coils
+
     def build(self):
         """ Return the payload buffer as a list
 

setup.py

@@ -87,6 +87,9 @@
             'pyasn1 >= 0.1.4',
             'pycrypto >= 2.6'
         ],
+        'tornado': [
+            'tornado >= 4.5.3'
+        ]
     },
     test_suite='nose.collector',
     cmdclass=command_classes,

test/test_payload.py

@@ -103,15 +103,24 @@ def testPayloadBuilderReset(self):
 
     def testPayloadBuilderWithRawPayload(self):
         """ Test basic bit message encoding/decoding """
-        builder = BinaryPayloadBuilder([b'\x12', b'\x34', b'\x56', b'\x78'], repack=True)
+        _coils1 = [False, True, True, True, True, False, False, False, False,
+                   True, False, True, False, True, True, False]
+        _coils2 = [False, True, False, True, False, True, True, False,
+                   False, True, True, True, True, False, False, False]
+
+        builder = BinaryPayloadBuilder([b'\x12', b'\x34', b'\x56', b'\x78'],
+                                       repack=True)
         self.assertEqual(b'\x12\x34\x56\x78', builder.to_string())
         self.assertEqual([13330, 30806], builder.to_registers())
 
+        self.assertEqual(_coils1, builder.to_coils())
+
         builder = BinaryPayloadBuilder([b'\x12', b'\x34', b'\x56', b'\x78'],
                                        byteorder=Endian.Big)
         self.assertEqual(b'\x12\x34\x56\x78', builder.to_string())
         self.assertEqual([4660, 22136], builder.to_registers())
         self.assertEqual('\x12\x34\x56\x78', str(builder))
+        self.assertEqual(_coils2, builder.to_coils())
 
     # ----------------------------------------------------------------------- #
     # Payload Decoder Tests