String and string array fixes

src/pyads/connection.py

@@ -76,6 +76,9 @@
     adsSyncDelDeviceNotificationReqEx,
     adsSyncSetTimeoutEx,
     ADSError,
+    get_value_from_ctype_data,
+    type_is_wstring,
+    type_is_string,
 )
 from .structs import (
     AmsAddr,
@@ -1019,9 +1022,9 @@ def parse_notification(
             addressof(contents) + SAdsNotificationHeader.data.offset
         )
         value: Any
-        if plc_datatype == PLCTYPE_STRING:
-            # read only until null-termination character
-            value = bytearray(data).split(b"\0", 1)[0].decode("utf-8")
+        if type_is_string(plc_datatype) or type_is_wstring(plc_datatype):
+            # Re-use string parsing from pyads_ex: (but doesn't work for other types)
+            value = get_value_from_ctype_data(data, plc_datatype)
 
         elif plc_datatype is not None and issubclass(plc_datatype, Structure):
             value = plc_datatype()

src/pyads/constants.py

@@ -31,10 +31,6 @@
 MAX_ADS_SUB_COMMANDS: int = 500
 
 
-class PLCTYPE_WSTRING:
-    """Special dummy class for handling WSTRING."""
-
-
 # plc data types:
 PLCTYPE_BOOL = c_bool
 PLCTYPE_BYTE = c_ubyte
@@ -45,6 +41,7 @@ class PLCTYPE_WSTRING:
 PLCTYPE_REAL = c_float
 PLCTYPE_SINT = c_int8
 PLCTYPE_STRING = c_char
+PLCTYPE_WSTRING = c_wchar
 PLCTYPE_TOD = c_int32
 PLCTYPE_UBYTE = c_ubyte
 PLCTYPE_UDINT = c_uint32

src/pyads/pyads_ex.py

@@ -245,6 +245,11 @@ def type_is_wstring(plc_type: Type) -> bool:
     if plc_type == PLCTYPE_WSTRING:
         return True
 
+    # If char array
+    if type(plc_type).__name__ == "PyCArrayType":
+        if plc_type._type_ == PLCTYPE_WSTRING:
+            return True
+
     return False
 
 
@@ -266,16 +271,15 @@ def get_value_from_ctype_data(read_data: Optional[Any], plc_type: Type) -> Any:
         return None
 
     if type_is_string(plc_type):
-        return read_data.value.decode("utf-8")
+        if hasattr(read_data, "value"):
+            return read_data.value.decode("utf-8")
+        return bytes(read_data).decode("utf-8").rstrip("\x00")
+        # `read_data.value` does not always exist, and without it all the null
+        # terminators needs to be removed after decoding
 
     if type_is_wstring(plc_type):
-        for ix in range(1, len(read_data), 2):
-            if (read_data[ix - 1], read_data[ix]) == (0, 0):
-                null_idx = ix - 1
-                break
-        else:
-            raise ValueError("No null-terminator found in buffer")
-        return bytearray(read_data[:null_idx]).decode("utf-16-le")
+        # `read_data.value` also exists, but could be wrong - explicitly decode instead:
+        return bytes(read_data).decode("utf-16-le").rstrip("\x00")
 
     if type(plc_type).__name__ == "PyCArrayType":
         return list(read_data)
@@ -842,12 +846,13 @@ def adsSyncReadReqEx2(
     index_group_c = ctypes.c_ulong(index_group)
     index_offset_c = ctypes.c_ulong(index_offset)
 
-    if type_is_string(data_type):
-        data = (STRING_BUFFER * PLCTYPE_STRING)()
-    elif type_is_wstring(data_type):
-        data = (STRING_BUFFER * ctypes.c_uint8)()
-    else:
-        data = data_type()
+    # Strings were handled specifically before, but their sizes are contained and we
+    # can proceed as normal
+    if data_type == PLCTYPE_STRING or data_type == PLCTYPE_WSTRING:
+        # This implies a string of size 1, which is so are we instead use a large fixed
+        # size buffer:
+        data_type = data_type * STRING_BUFFER
+    data = data_type()
 
     data_pointer = ctypes.pointer(data)
     data_length = ctypes.c_ulong(ctypes.sizeof(data))
@@ -868,11 +873,11 @@ def adsSyncReadReqEx2(
     if error_code:
         raise ADSError(error_code)
 
-    # If we're reading a value of predetermined size (anything but a string or wstring),
-    # validate that the correct number of bytes were read
+    # If we're reading a value of predetermined size (anything but strings, which can be shorted
+    # because of null-termination), validate that the correct number of bytes were read
     if (
-        check_length
-        and not(type_is_string(data_type) or type_is_wstring(data_type))
+        check_length 
+        and not (type_is_string(data_type) or type_is_wstring(data_type))
         and bytes_read.value != data_length.value
     ):
         raise RuntimeError(

src/pyads/symbol.py

@@ -302,9 +302,7 @@ def get_type_from_str(type_str: str) -> Optional[Type[PLCDataType]]:
         # If simple scalar
         plc_name = "PLCTYPE_" + type_str
 
-        # if type is WSTRING just return the PLCTYPE constant
-        if plc_name.startswith("PLCTYPE_WSTRING"):
-            return constants.PLCTYPE_WSTRING
+        # WSTRING used to be captured specifically but is now handled as normal
 
         if hasattr(constants, plc_name):
             # Map e.g. 'LREAL' to 'PLCTYPE_LREAL' directly based on the name
@@ -348,6 +346,10 @@ def get_type_from_str(type_str: str) -> Optional[Type[PLCDataType]]:
             scalar_type = AdsSymbol.get_type_from_str(scalar_type_str)
 
             if scalar_type:
+                if scalar_type in [constants.PLCTYPE_STRING, constants.PLCTYPE_WSTRING]:
+                    # E.g. `STRING(80)` actually has a size of 81 including the string
+                    # terminator, so add one to the size
+                    size = size + 1
                 return scalar_type * size
 
         # We allow unmapped types at this point - Instead we will throw  an

tests/test_connection_class.py

@@ -1452,7 +1452,7 @@ def test_read_wstring(self):
         var = PLCVariable(
             "wstr",
             expected1.encode("utf-16-le") + b"\x00\x00",
-            constants.ADST_WSTRING, f"WSTRING({len(expected1)})"
+            constants.ADST_WSTRING, "WSTRING(80)"
         )
         self.handler.add_variable(var)
 
@@ -1490,9 +1490,9 @@ def test_read_write_list_wstr_array(self):
 
         # Add to test plc
         self.handler.add_variable(PLCVariable(
-            name = "wstr_test_array", 
-            value = bytes(w_string_bytes), 
-            ads_type = constants.ADST_WSTRING, 
+            name = "wstr_test_array",
+            value = bytes(w_string_bytes),
+            ads_type = constants.ADST_WSTRING,
             symbol_type = f"WSTRING({w_string_char_size})"))
 
 
@@ -1540,9 +1540,9 @@ def test_read_write_list_str_array(self):
 
         # Add to test plc
         self.handler.add_variable(PLCVariable(
-            name = "str_test_array", 
-            value = bytes(string_bytes), 
-            ads_type = constants.ADST_STRING, 
+            name = "str_test_array",
+            value = bytes(string_bytes),
+            ads_type = constants.ADST_STRING,
             symbol_type = f"STRING({string_char_size})"))
 
 
@@ -1591,9 +1591,9 @@ def test_read_write_list_int_array(self):
 
         # Add to test plc
         self.handler.add_variable(PLCVariable(
-            name = "int_test_array", 
-            value = bytes(int_array_bytes), 
-            ads_type = constants.ADST_INT16, 
+            name = "int_test_array",
+            value = bytes(int_array_bytes),
+            ads_type = constants.ADST_INT16,
             symbol_type = f"INT"))
 
 
@@ -1643,17 +1643,17 @@ def test_read_write_list_real_array(self):
 
         # Add to test plc
         self.handler.add_variable(PLCVariable(
-            name = "real_test_array", 
-            value = bytes(real_array_bytes), 
-            ads_type = constants.ADST_REAL32, 
+            name = "real_test_array",
+            value = bytes(real_array_bytes),
+            ads_type = constants.ADST_REAL32,
             symbol_type = f"REAL"))
 
 
         # Read variable
         with self.plc:
             read_values = self.plc.read_list_by_name(["real_test_array"])
 
-        # Verify result to 1dp 
+        # Verify result to 1dp
         for i, value in enumerate(read_values["real_test_array"]):
             self.assertEqual(expected_real_array[i], round(value, 1))
 
@@ -1669,7 +1669,7 @@ def test_read_write_list_real_array(self):
         with self.plc:
             read_values = self.plc.read_list_by_name(["real_test_array"])
 
-        # Verify result to 1dp 
+        # Verify result to 1dp
         for i, value in enumerate(read_values["real_test_array"]):
             self.assertEqual(expected_real_array[i], round(value, 1))
 

tests/test_symbol.py

@@ -643,7 +643,7 @@ def test_arrays(self):
     def test_string(self):
         type_str = 'STRING(80)'  # This is how a string might appear
         plc_type = AdsSymbol.get_type_from_str(type_str)
-        self.assertSizeOf(plc_type, 1 * 80)
+        self.assertSizeOf(plc_type, 1 * 81)
 
 
 if __name__ == "__main__":