diff --git a/src/pyads/connection.py b/src/pyads/connection.py index d6f5ce40..dd7890aa 100644 --- a/src/pyads/connection.py +++ b/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() diff --git a/src/pyads/constants.py b/src/pyads/constants.py index 875e6d8d..f5a5eebd 100644 --- a/src/pyads/constants.py +++ b/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 diff --git a/src/pyads/pyads_ex.py b/src/pyads/pyads_ex.py index 11c79acc..89869f4d 100644 --- a/src/pyads/pyads_ex.py +++ b/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( diff --git a/src/pyads/symbol.py b/src/pyads/symbol.py index 3ea6566b..115002ed 100644 --- a/src/pyads/symbol.py +++ b/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 diff --git a/tests/test_connection_class.py b/tests/test_connection_class.py index 9e7a3570..5d03494d 100644 --- a/tests/test_connection_class.py +++ b/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,9 +1643,9 @@ 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")) @@ -1653,7 +1653,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)) @@ -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)) diff --git a/tests/test_symbol.py b/tests/test_symbol.py index 02cb3a91..b1f5b490 100644 --- a/tests/test_symbol.py +++ b/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__":