TestModbusProtocol test_client_server_count passes

Regression to pass also write a coil to False

Author: francozappa

minicps/devices.py

@@ -227,9 +227,12 @@ def get(self, what):
         else:
             return self._state._get(what)
 
-    def send(self, what, value, address):
+    def send(self, what, value, address, **kwargs):
         """Send (write) a value to another network host.
 
+        ``kwargs`` dict is used to pass extra key-value pair according to the
+        used protocol.
+
         :param tuple what: field[s] identifier[s]
         :param value: value to be setted
         :param str address: ``ip[:port]``
@@ -245,6 +248,9 @@ def send(self, what, value, address):
     def recieve(self, what, address, **kwargs):
         """Receive (read) a value from another network host.
 
+        ``kwargs`` dict is used to pass extra key-value pair according to the
+        used protocol.
+
         :param tuple what: field[s] identifier[s]
         :param str address: ``ip[:port]``
 

minicps/protocols.py

@@ -86,7 +86,7 @@ def _stop_server(cls, address):
 
         print '_stop_server: please override me.'
 
-    def _send(self, what, value, address):
+    def _send(self, what, value, address, **kwargs):
         """Send (write) a value to another host.
 
         :what: to send
@@ -349,7 +349,7 @@ def _stop_server(cls, server):
         except Exception as error:
             print 'ERROR stop enip server: ', error
 
-    def _send(self, what, value, address='localhost:44818'):
+    def _send(self, what, value, address='localhost:44818', **kwargs):
         """Send (write) a value to another host.
 
         It is a blocking operation the parent process will wait till the child
@@ -607,14 +607,20 @@ def _stop_server(cls, server):
             print 'ERROR stop modbus server: ', error
 
 
-    def _send(self, what, value, address='localhost:502'):
+    def _send(self, what, value, address='localhost:502', **kwargs):
         """Send (write) a value to another host.
 
         It is a blocking operation the parent process will wait till the child
         cpppo process returns.
 
+        Boolean values are converted back and forth to integers because
+        ``argparse`` does not handle bool arguments correctly, eg: False
+        string is evaluated and passed as True.
+
         pymodbus has a ``count`` kwarg to perform sequential read and write
-        starting from the offset passed inside ``what``.
+        starting from the offset passed inside ``what``. Default call will
+        write one value at a time. Pass a list of values if ``count`` is
+        greater than one.
 
 
         :what: tuple addressing what
@@ -629,10 +635,40 @@ def _send(self, what, value, address='localhost:502'):
         MODE = '-m {} '.format('w')
         TYPE = '-t {} '.format(what[0])
         OFFSET = '-o {} '.format(what[1])  # NOTE: 0-based
+
+        # NOTE: value is a list of bools or ints when write multiple times
+        if 'count' in kwargs and kwargs['count'] > 1:
+            count = kwargs['count']
+            COUNT = '--count {} '.format(count)
+        else:
+            count = 1
+            COUNT = '--count {} '.format(count)
+
+        # NOTE: value is a int when writing to a register
         if what[0] == 'HR':
-            VALUE = '-r {} '.format(value)
+            if count == 1:
+                VALUE = '-r {} '.format(value)
+            else:
+                VALUE = '-r '
+                for v in value:
+                    VALUE += str(v)
+                    VALUE += ' '
+
+        # NOTE: value is a bool when writing to a coil
         elif what[0] == 'CO':
-            VALUE = '-c {} '.format(value)
+            if count == 1:
+                if value == True:
+                    VALUE = '-c {} '.format(1)
+                else:
+                    VALUE = '-c {} '.format(0)
+            else:
+                VALUE = '-c '
+                for v in value:
+                    if v ==  True:
+                        VALUE += str(1)
+                    else:
+                        VALUE += str(0)
+                    VALUE += ' '
         else:
             raise ValueError('IR and DI are read only data.')
 
@@ -644,6 +680,7 @@ def _send(self, what, value, address='localhost:502'):
             MODE +
             TYPE +
             OFFSET +
+            COUNT +
             VALUE
         )
         # print 'DEBUG modbus_send cmd shlex list: ', cmd
@@ -664,7 +701,7 @@ def _receive(self, what, address='localhost:502', **kwargs):
         cpppo process returns.
 
         pymodbus has a ``count`` kwarg to perform sequential read and write
-        starting from the offset passed inside ``what``. Default setup will
+        starting from the offset passed inside ``what``. Default call will
         request and return one value at a time.
 
         The return type depends on the request type:

scripts/pymodbus/synch-client.py

@@ -40,11 +40,12 @@
             type=int, choices=range(1,2000),  # NOTE: bounds from the standard
             default=1)
     parser.add_argument('-r', dest='register',
-            help='register value', type=int, choices=range(0, 65536),
+            help='list of int values', type=int,
+            choices=range(0, 65536), nargs='+',
             default=0)
     parser.add_argument('-c', dest='coil',
-            help='coil value', type=bool,
-            default=True)  # NOTE: implicit True False choice
+            help='list of 0 (False) or 1 (True) int values', type=int, nargs='+',
+            default=0, choices=[0, 1])  #  argparse does not manage bool well
 
     args = parser.parse_args()
 
@@ -65,16 +66,31 @@
         # NOTE: write_register
         if args.type == 'HR':
             if args.count == 1:
-                hr_write = client.write_register(args.offset, int(args.register))
+                hr_write = client.write_register(args.offset, args.register[0])
                 assert(hr_write.function_code < 0x80)
-            # TODO: implement
-            # else:
-            #     hrs_write = client.write_registers(args.offset, args.values)
+            else:
+                hrs_write = client.write_registers(args.offset, args.register)
+                assert(hrs_write.function_code < 0x80)
 
-        # NOTE: write_coil
+        # NOTE: write_coil: map integers to bools
         elif args.type == 'CO':
-            co_write = client.write_coil(args.offset, args.coil)
-            assert(co_write.function_code < 0x80)
+
+            if args.count == 1:
+                if args.coil == 1:
+                    co_write = client.write_coil(args.offset, True)
+                else:
+                    co_write = client.write_coil(args.offset, False)
+                assert(co_write.function_code < 0x80)
+
+            else:
+                coils = []
+                for c in args.coil:
+                    if c == 1:
+                        coils.append(True)
+                    else:
+                        coils.append(False)
+                cos_write = client.write_coils(args.offset, coils)
+                assert(cos_write.function_code < 0x80)
 
 
     elif args.mode == 'r':

tests/protocols_tests.py

@@ -347,6 +347,7 @@ def test_receive(self):
             print 'ERROR test_receive: ', error
             assert False
 
+    @SkipTest
     def test_client_server(self):
 
         ADDRESS = 'localhost:502'
@@ -363,6 +364,12 @@ def test_client_server(self):
             modbus._send(what, value, ADDRESS)
             what = ('CO', 0)
             eq_(modbus._receive(what, ADDRESS), True)
+
+            what = ('CO', 1)
+            value = False
+            modbus._send(what, value, ADDRESS)
+            what = ('CO', 1)
+            eq_(modbus._receive(what, ADDRESS), False)
             print('')
 
             print('TEST: Write and read holding registers')
@@ -398,7 +405,6 @@ def test_receive_count(self):
 
         ADDRESS = 'localhost:502'
         TAGS = (20, 20, 20, 20)
-        OFFSET = 10
 
         try:
             server = ModbusProtocol._start_server(ADDRESS, TAGS)
@@ -430,4 +436,31 @@ def test_receive_count(self):
             ModbusProtocol._stop_server(server)
             print 'ERROR test_receive_count: ', error
             assert False
+
+    def test_client_server_count(self):
+
+        client = ModbusProtocol(
+            protocol=TestModbusProtocol.CLIENT_PROTOCOL)
+
+        ADDRESS = 'localhost:502'
+        TAGS = (50, 50, 50, 50)
+
+        try:
+            server = ModbusProtocol._start_server(ADDRESS, TAGS)
+            time.sleep(1.0)
+
+            print('TEST: Write and Read holding registers, offset=4, count=3')
+            what = ('HR', 4)
+            values = [1, 2, 3]
+            client._send(what, values, ADDRESS, count=3)
+            eq_(client._receive(what, ADDRESS, count=3), [1, 2, 3])
+            print('')
+
+
+            ModbusProtocol._stop_server(server)
+
+        except Exception as error:
+            ModbusProtocol._stop_server(server)
+            print 'ERROR test_client_server_count: ', error
+            assert False
 # }}}