Merge pull request #50 from ottowayi/develop

Develop

pycomm3/__init__.py

@@ -24,7 +24,7 @@
 # SOFTWARE.
 #
 
-__version_info__ = (0, 6, 6)
+__version_info__ = (0, 7, 0)
 __version__ = '.'.join(f'{x}' for x in __version_info__)
 
 from typing import NamedTuple, Any, Optional

pycomm3/clx.py

@@ -29,6 +29,7 @@
 import logging
 import datetime
 import time
+import itertools
 from functools import wraps
 from os import urandom
 from typing import Union, List, Sequence, Tuple, Optional
@@ -169,7 +170,6 @@ def __init__(self, path: str, *args,  large_packets: bool = True, micro800: bool
             if init_info:
                 self._micro800 = self._list_identity().startswith(MICRO800_PREFIX)
                 self.get_plc_info()
-                # self._micro800 = self.info['device_type'].startswith(MICRO800_PREFIX)
                 _, _path = _parse_connection_path(path, self._micro800)  # need to update path if using a Micro800
                 self.attribs['cip_path'] = _path
                 self.use_instance_ids = (self.info.get('version_major', 0) >= MIN_VER_INSTANCE_IDS) and not self._micro800
@@ -1554,8 +1554,8 @@ def _writable_value_structure(value, elements, data_type):
         return _pack_structure(value, data_type)
 
 
-def _pack_string(value, string_len):
-    sint_array = [b'\x00' for _ in range(string_len)]
+def _pack_string(value, string_len, struct_size):
+    sint_array = [b'\x00' for _ in range(struct_size-4)]  # 4 for .LEN
     if len(value) > string_len:
         value = value[:string_len]
     for i, s in enumerate(value):
@@ -1564,17 +1564,53 @@ def _pack_string(value, string_len):
     return pack_dint(len(value)) + b''.join(sint_array)
 
 
-def _pack_structure(val, data_type):
+def _pack_structure(value, data_type):
     string_len = data_type.get('string')
-    if string_len is None:
-        raise NotImplementedError('Writing of structures besides strings is not supported')
+    # if string_len is None:
+    #     raise NotImplementedError('Writing of structures besides strings is not supported')
 
     if string_len:
-        packed_bytes = _pack_string(val, string_len)
+        data = _pack_string(value, string_len, data_type['template']['structure_size'])
     else:
-        packed_bytes = b''  # TODO: support for structure writing here
+        data = [0 for _ in range(data_type['template']['structure_size'])]
+        try:
+            # NOTE:  start with bytes(object-definition-size) , then replace sections with offset + data len
+            for val, attr in zip(value, data_type['attributes']):
+                dtype = data_type['internal_tags'][attr]
+                offset = dtype['offset']
+
+                ary = dtype.get('array')
+                if dtype['tag_type'] == 'struct':
+                    if ary:
+                        value_bytes = [_pack_structure(val[i], dtype['data_type']) for i in range(ary)]
+                    else:
+                        value_bytes = [_pack_structure(val, dtype['data_type']), ]
+                else:
+                    pack_func = PACK_DATA_FUNCTION[dtype['data_type']]
+                    bit = dtype.get('bit')
+                    if bit is not None:
+                        if val:
+                            data[offset] |= 1 << bit
+                        else:
+                            data[offset] &= ~(1 << bit)
+                        continue
+
+                    if ary:
+                        value_bytes = [pack_func(val[i]) for i in range(ary)]
+                    else:
+                        value_bytes = [pack_func(val), ]
+
+                val_bytes = list(itertools.chain.from_iterable(value_bytes))
+                data[offset:offset+len(val_bytes)] = val_bytes
+
+        except Exception as err:
+            raise RequestError('Value Invalid for Structure', err)
+
+    return bytes(data)
+
 
-    return packed_bytes + b'\x00' * (len(packed_bytes) % 4)  # pad data to 4-byte boundaries
+def _pad(data):
+    return data + b'\x00' * (len(data) % 4)  # pad data to 4-byte boundaries
 
 
 def _bit_request(tag_data, bit_requests):

pycomm3/packets/responses.py

@@ -453,20 +453,20 @@ def parse_read_reply_struct(data, data_type):
 
             values[tag] = value
         elif datatype.get('string'):
-            str_size = datatype.get('string') + 4
-            str_size += str_size % 4
+            str_size = datatype['template']['structure_size']
             if array:
                 array_data = data[offset:offset + (str_size * array)]
                 values[tag] = [parse_string(array_data[i:i+str_size]) for i in range(0, len(array_data), str_size)]
             else:
                 values[tag] = parse_string(data[offset:offset + str_size])
         else:
+            struct_size = datatype['template']['structure_size']
             if array:
-                ary_data = data[offset:offset + (size * array)]
-                values[tag] = [parse_read_reply_struct(ary_data[i:i + size], datatype) for i in
-                               range(0, len(ary_data), size)]
+                ary_data = data[offset:offset + (struct_size * array)]
+                values[tag] = [parse_read_reply_struct(ary_data[i:i + struct_size], datatype) for i in
+                               range(0, len(ary_data), struct_size)]
             else:
-                values[tag] = parse_read_reply_struct(data[offset:offset + size], datatype)
+                values[tag] = parse_read_reply_struct(data[offset:offset + struct_size], datatype)
 
     return {k: v for k, v in values.items() if k in data_type['attributes']}
 

tests/test_writes.py

@@ -2,6 +2,7 @@
 import pytest
 from itertools import chain
 from . import tag_only
+import time
 
 atomic_tests = [  # (tag name, data type, value)
 
@@ -56,3 +57,31 @@ def test_atomic_writes(plc, tag_name, data_type, value):
     assert result.error is None
     assert result.tag == tag_only(tag_name)
     assert result.type == data_type
+
+
+def test_struct_write(plc):
+    values = (
+        1, [2, 3, 4], 'Hello World!!!', True, 123.45, True,
+        [True, 1, 2, 3, 4, 5],
+        [
+            [True, 1, 1, 1, 1, 1],
+            [True, 2, 2, 2, 2, 2],
+            [True, 3, 3, 3, 3, 3]
+        ]
+    )
+
+    plc.write(('TestStructWrite', 1))
+    time.sleep(1)  # in case scan rate is too slow to for FAL to complete before write
+    plc.write(('TestUDT1_1', values))
+    tag = plc.read('TestUDT1_1')
+
+    assert tag.value['bool1'] is True
+    assert tag.value['bool2'] is True
+    assert tag.value['dint'] == 1
+    assert tag.value['ints'] == [2, 3, 4]
+    assert isclose(tag.value['real'], 123.45, rel_tol=1e-4)
+    assert tag.value['string'] == 'Hello World!!!'
+    assert tag.value['udt'] == {'bool': True, 'dint': 3, 'int': 2, 'real': 4.0, 'sint': 1}
+    assert tag.value['udts'] == [{'bool': True, 'dint': 1, 'int': 1, 'real': 1.0, 'sint': 1},
+                                 {'bool': True, 'dint': 2, 'int': 2, 'real': 2.0, 'sint': 2},
+                                 {'bool': True, 'dint': 3, 'int': 3, 'real': 3.0, 'sint': 3}]