TestModbusProtocol read HR and IR passes

Author: francozappa

minicps/protocols.py

@@ -404,9 +404,7 @@ def _receive(self, what, address='localhost:44818'):
         # print 'DEBUG enip _receive cmd shlex list: ', cmd
 
         try:
-            client = subprocess.Popen(
-                cmd,
-                shell=False,
+            client = subprocess.Popen(cmd, shell=False,
                 stdout=subprocess.PIPE)
 
             # client.communicate is blocking
@@ -417,6 +415,7 @@ def _receive(self, what, address='localhost:44818'):
             # between a pair of square brackets
             raw_string = raw_out[0]
             out = raw_string[(raw_string.find('[') + 1):raw_string.find(']')]
+
             return out
 
         except Exception as error:
@@ -655,8 +654,7 @@ def _send(self, what, value, address='localhost:502'):
             print 'ERROR modbus _send: ', error
 
 
-    # TODO: implement it
-    def _receive(self, what, address='localhost:44818'):
+    def _receive(self, what, address='localhost:502'):
         """Receive (read) a value from another host.
 
         It is a blocking operation the parent process will wait till the child
@@ -667,35 +665,46 @@ def _receive(self, what, address='localhost:44818'):
 
         :returns: tag value as a `str`
         """
-
-        tag_string = ''
-        tag_string = EnipProtocol._tuple_to_cpppo_tag(what)
+        colon_index = address.find(':')
+        IP = '-i {} '.format(address[:colon_index])
+        PORT = '-p {} '.format(address[colon_index+1:])
+        # NOTE: following data is validated by client script
+        MODE = '-m {} '.format('r')
+        TYPE = '-t {} '.format(what[0])
+        OFFSET = '-o {} '.format(what[1])  # NOTE: 0-based
 
         cmd = shlex.split(
             self._client_cmd +
-            '--log ' + self._client_log +
-            '--address ' + address +
-            ' ' + tag_string
+            IP +
+            PORT +
+            MODE +
+            TYPE +
+            OFFSET
         )
-        # print 'DEBUG enip _receive cmd shlex list: ', cmd
+        print 'DEBUG modbus_receive cmd shlex list: ', cmd
 
         try:
-            client = subprocess.Popen(
-                cmd,
-                shell=False,
+            client = subprocess.Popen(cmd,shell=False,
                 stdout=subprocess.PIPE)
 
             # client.communicate is blocking
             raw_out = client.communicate()
-            # print 'DEBUG enip _receive raw_out: ', raw_out
+            print 'DEBUG modbus _receive raw_out: ', raw_out
 
             # value is stored as first tuple element
             # between a pair of square brackets
-            raw_string = raw_out[0]
-            out = raw_string[(raw_string.find('[') + 1):raw_string.find(']')]
+            raw_string = raw_out[0].strip()
+
+            # NOTE: registers store int
+            if what[0] == 'HR' or what[0] == 'IR':
+                out = int(raw_string)
+            # NOTE: coils and discrete inputs store bool
+            elif what[0] == 'CO' or what[0] == 'DI':
+                pass
+
             return out
 
         except Exception as error:
-            print 'ERROR enip _receive: ', error
+            print 'ERROR modbus _receive: ', error
 
 # }}}

scripts/pymodbus/synch-client.py

@@ -54,6 +54,7 @@
 
     client.connect()
 
+    # TODO: check if asserts are slowing down read/write
     if args.mode == 'w':
 
         # NOTE: write_register
@@ -66,4 +67,21 @@
             co_write = client.write_coil(args.offset, args.coil)
             assert(co_write.function_code < 0x80)
 
+
+    elif args.mode == 'r':
+
+        # NOTE: read_holding_registers
+        if args.type == 'HR':
+            hr_read = client.read_holding_registers(args.offset, count=1)
+            assert(hr_read.function_code < 0x80)
+            print(hr_read.registers[0])
+
+        # NOTE: read_holding_registers
+        elif args.type == 'IR':
+            ir_read = client.read_input_registers(args.offset, count=1)
+            assert(ir_read.function_code < 0x80)
+            print(ir_read.registers[0])
+
+
+
     client.close()

tests/protocols_tests.py

@@ -278,6 +278,7 @@ def test_init_server(self):
             print 'ERROR test_init_server: ', error
 
 
+    @SkipTest
     def test_send(self):
 
         client = ModbusProtocol(
@@ -311,5 +312,41 @@ def test_send(self):
             ModbusProtocol._stop_server(server)
             print 'ERROR test_send: ', error
 
+    def test_receive(self):
+
+        client = ModbusProtocol(
+            protocol=CLIENT_PROTOCOL)
+
+        ADDRESS = 'localhost:502'
+        TAGS = (20, 20, 20, 20)
+        OFFSET = 10
+
+        try:
+            server = ModbusProtocol._start_server(ADDRESS, TAGS)
+            time.sleep(1.0)
+
+            print('TEST: Read to holding registers')
+            for count in range(0, OFFSET):
+                what = ('HR', count)
+                eq_(client._receive(what, ADDRESS), 0)
+            print('')
+
+            print('TEST: Read to input registers')
+            for count in range(0, OFFSET):
+                what = ('IR', count)
+                eq_(client._receive(what, ADDRESS), 0)
+            print('')
+
+            # print('TEST: Read to coils')
+            # for count in range(0, OFFSET):
+            #     what = ('CO', count)
+            #     client._receive(what, ADDRESS)
+            # print('')
+
+            ModbusProtocol._stop_server(server)
+
+        except Exception as error:
+            ModbusProtocol._stop_server(server)
+            print 'ERROR test_receive: ', error
 
 # }}}