From 3df551aa93e63a67f120915bccc4a5b44160c2dd Mon Sep 17 00:00:00 2001 From: Thiago Alves Date: Wed, 28 Jan 2026 19:07:00 -0500 Subject: [PATCH 1/2] fix: Add missing IEC 61131-3 type mappings to OPC-UA plugin SINT variables were not mapped in map_plc_to_opcua_type(), causing BadInternalError when OPC-UA clients tried to read them. The asyncua library failed to serialize responses because unmapped types defaulted to ua.VariantType.Variant, which couldn't handle raw integer values. Added type mappings for: - SINT (SByte), USINT (Byte) - UINT (UInt16), WORD (UInt16) - UDINT (UInt32), DWORD (UInt32) - ULINT (UInt64), LWORD (UInt64) - LREAL (Double) Also added corresponding value conversion handling in both convert_value_for_opcua() and convert_value_for_plc() functions to support both read and write operations. Co-Authored-By: Claude Opus 4.5 --- .../plugins/python/opcua/opcua_utils.py | 117 ++++++++++++++---- .../plugins/opcua/test_type_conversions.py | 69 +++++++++++ 2 files changed, 165 insertions(+), 21 deletions(-) diff --git a/core/src/drivers/plugins/python/opcua/opcua_utils.py b/core/src/drivers/plugins/python/opcua/opcua_utils.py index 6b0c9eea..41748a56 100644 --- a/core/src/drivers/plugins/python/opcua/opcua_utils.py +++ b/core/src/drivers/plugins/python/opcua/opcua_utils.py @@ -27,18 +27,33 @@ def map_plc_to_opcua_type(plc_type: str) -> ua.VariantType: """Map plc datatype to OPC-UA VariantType.""" type_mapping = { "BOOL": ua.VariantType.Boolean, - "BYTE": ua.VariantType.Byte, - "INT": ua.VariantType.Int16, - "INT32": ua.VariantType.Int32, - "DINT": ua.VariantType.Int32, - "LINT": ua.VariantType.Int64, + # 8-bit types + "SINT": ua.VariantType.SByte, # Signed 8-bit integer + "USINT": ua.VariantType.Byte, # Unsigned 8-bit integer + "BYTE": ua.VariantType.Byte, # Unsigned 8-bit (alias for USINT) + # 16-bit types + "INT": ua.VariantType.Int16, # Signed 16-bit integer + "UINT": ua.VariantType.UInt16, # Unsigned 16-bit integer + "WORD": ua.VariantType.UInt16, # Unsigned 16-bit (bit string) + # 32-bit types + "DINT": ua.VariantType.Int32, # Signed 32-bit integer + "INT32": ua.VariantType.Int32, # Alias for DINT + "UDINT": ua.VariantType.UInt32, # Unsigned 32-bit integer + "DWORD": ua.VariantType.UInt32, # Unsigned 32-bit (bit string) + # 64-bit types + "LINT": ua.VariantType.Int64, # Signed 64-bit integer + "ULINT": ua.VariantType.UInt64, # Unsigned 64-bit integer + "LWORD": ua.VariantType.UInt64, # Unsigned 64-bit (bit string) + # Floating point types "FLOAT": ua.VariantType.Float, - "REAL": ua.VariantType.Float, # IEC 61131-3 REAL = 32-bit float + "REAL": ua.VariantType.Float, # IEC 61131-3 REAL = 32-bit float + "LREAL": ua.VariantType.Double, # IEC 61131-3 LREAL = 64-bit float + # String type "STRING": ua.VariantType.String, # TIME-related types - "TIME": ua.VariantType.Int64, # Duration in milliseconds - "TOD": ua.VariantType.DateTime, # Time of day as DateTime (current date + time) - "DATE": ua.VariantType.DateTime, # Date as DateTime (date only, time set to 00:00:00) + "TIME": ua.VariantType.Int64, # Duration in milliseconds + "TOD": ua.VariantType.DateTime, # Time of day as DateTime (current date + time) + "DATE": ua.VariantType.DateTime, # Date as DateTime (date only, time set to 00:00:00) "DT": ua.VariantType.DateTime, # Date and Time as OPC-UA DateTime } mapped_type = type_mapping.get(plc_type.upper(), ua.VariantType.Variant) @@ -88,10 +103,20 @@ def convert_value_for_opcua(datatype: str, value: Any) -> Any: else: return bool(value) - elif datatype.upper() in ["BYTE", "Byte"]: + elif datatype.upper() in ["SINT", "Sint"]: + # Ensure proper int8 type for OPC-UA compatibility (signed 8-bit) + clamped_value = max(-128, min(127, int(value))) + return ctypes.c_int8(clamped_value).value + + elif datatype.upper() in ["BYTE", "Byte", "USINT", "Usint"]: # Ensure proper uint8 type for OPC-UA compatibility return ctypes.c_uint8(max(0, min(255, int(value)))).value - + + elif datatype.upper() in ["UINT", "Uint", "WORD", "Word"]: + # Ensure proper uint16 type for OPC-UA compatibility + clamped_value = max(0, min(65535, int(value))) + return ctypes.c_uint16(clamped_value).value + elif datatype.upper() in ["INT", "Int"]: # Ensure proper int16 type - critical for OPC-UA compatibility clamped_value = max(-32768, min(32767, int(value))) @@ -101,10 +126,20 @@ def convert_value_for_opcua(datatype: str, value: Any) -> Any: # Ensure proper int32 type for OPC-UA compatibility clamped_value = max(-2147483648, min(2147483647, int(value))) return ctypes.c_int32(clamped_value).value - + + elif datatype.upper() in ["UDINT", "Udint", "DWORD", "Dword"]: + # Ensure proper uint32 type for OPC-UA compatibility + clamped_value = max(0, min(4294967295, int(value))) + return ctypes.c_uint32(clamped_value).value + elif datatype.upper() in ["LINT", "Lint"]: return int(value) # int64 - + + elif datatype.upper() in ["ULINT", "Ulint", "LWORD", "Lword"]: + # Ensure proper uint64 type for OPC-UA compatibility + clamped_value = max(0, min(18446744073709551615, int(value))) + return ctypes.c_uint64(clamped_value).value + elif datatype.upper() in ["FLOAT", "REAL"]: # Float/Real values are stored as integers in debug variables # Convert back to float if it's an integer representation @@ -114,7 +149,17 @@ def convert_value_for_opcua(datatype: str, value: Any) -> Any: except: return float(value) return float(value) - + + elif datatype.upper() in ["LREAL", "Lreal"]: + # LREAL (64-bit float) values are stored as integers in debug variables + # Convert back to double if it's an integer representation + if isinstance(value, int): + try: + return struct.unpack('d', struct.pack('Q', value))[0] + except: + return float(value) + return float(value) + elif datatype.upper() in ["STRING", "String"]: return str(value) @@ -230,23 +275,43 @@ def convert_value_for_plc(datatype: str, value: Any) -> Any: else: return int(bool(value)) - elif datatype.upper() in ["BYTE", "Byte"]: + elif datatype.upper() in ["SINT", "Sint"]: + # Ensure proper int8 type for PLC compatibility (signed 8-bit) + clamped_value = max(-128, min(127, int(value))) + return ctypes.c_int8(clamped_value).value + + elif datatype.upper() in ["BYTE", "Byte", "USINT", "Usint"]: # Ensure proper uint8 type for PLC compatibility return ctypes.c_uint8(max(0, min(255, int(value)))).value - + elif datatype.upper() in ["INT", "Int"]: - # Ensure proper int16 type for PLC compatibility + # Ensure proper int16 type for PLC compatibility clamped_value = max(-32768, min(32767, int(value))) return ctypes.c_int16(clamped_value).value - + + elif datatype.upper() in ["UINT", "Uint", "WORD", "Word"]: + # Ensure proper uint16 type for PLC compatibility + clamped_value = max(0, min(65535, int(value))) + return ctypes.c_uint16(clamped_value).value + elif datatype.upper() in ["DINT", "Dint", "INT32", "Int32"]: # Ensure proper int32 type for PLC compatibility clamped_value = max(-2147483648, min(2147483647, int(value))) return ctypes.c_int32(clamped_value).value - + + elif datatype.upper() in ["UDINT", "Udint", "DWORD", "Dword"]: + # Ensure proper uint32 type for PLC compatibility + clamped_value = max(0, min(4294967295, int(value))) + return ctypes.c_uint32(clamped_value).value + elif datatype.upper() in ["LINT", "Lint"]: return int(value) # int64 - + + elif datatype.upper() in ["ULINT", "Ulint", "LWORD", "Lword"]: + # Ensure proper uint64 type for PLC compatibility + clamped_value = max(0, min(18446744073709551615, int(value))) + return ctypes.c_uint64(clamped_value).value + elif datatype.upper() in ["FLOAT", "REAL"]: # Convert float to int representation for storage if isinstance(value, float): @@ -256,7 +321,17 @@ def convert_value_for_plc(datatype: str, value: Any) -> Any: return int(value) else: return int(float(value)) - + + elif datatype.upper() in ["LREAL", "Lreal"]: + # Convert double to int representation for storage (64-bit) + if isinstance(value, float): + try: + return struct.unpack('Q', struct.pack('d', value))[0] + except: + return int(value) + else: + return int(float(value)) + elif datatype.upper() in ["STRING", "String"]: return str(value) diff --git a/tests/pytest/plugins/opcua/test_type_conversions.py b/tests/pytest/plugins/opcua/test_type_conversions.py index 1e2a730c..ef893ba6 100644 --- a/tests/pytest/plugins/opcua/test_type_conversions.py +++ b/tests/pytest/plugins/opcua/test_type_conversions.py @@ -38,11 +38,31 @@ def test_bool_mapping(self): assert map_plc_to_opcua_type("bool") == ua.VariantType.Boolean assert map_plc_to_opcua_type("Bool") == ua.VariantType.Boolean + def test_sint_mapping(self): + """SINT should map to SByte (signed 8-bit integer).""" + assert map_plc_to_opcua_type("SINT") == ua.VariantType.SByte + assert map_plc_to_opcua_type("sint") == ua.VariantType.SByte + + def test_usint_mapping(self): + """USINT should map to Byte (unsigned 8-bit integer).""" + assert map_plc_to_opcua_type("USINT") == ua.VariantType.Byte + assert map_plc_to_opcua_type("usint") == ua.VariantType.Byte + def test_byte_mapping(self): """BYTE should map to Byte.""" assert map_plc_to_opcua_type("BYTE") == ua.VariantType.Byte assert map_plc_to_opcua_type("byte") == ua.VariantType.Byte + def test_uint_mapping(self): + """UINT should map to UInt16 (unsigned 16-bit integer).""" + assert map_plc_to_opcua_type("UINT") == ua.VariantType.UInt16 + assert map_plc_to_opcua_type("uint") == ua.VariantType.UInt16 + + def test_word_mapping(self): + """WORD should map to UInt16.""" + assert map_plc_to_opcua_type("WORD") == ua.VariantType.UInt16 + assert map_plc_to_opcua_type("word") == ua.VariantType.UInt16 + def test_int_mapping(self): """INT should map to Int16.""" assert map_plc_to_opcua_type("INT") == ua.VariantType.Int16 @@ -58,11 +78,31 @@ def test_int32_mapping(self): assert map_plc_to_opcua_type("INT32") == ua.VariantType.Int32 assert map_plc_to_opcua_type("int32") == ua.VariantType.Int32 + def test_udint_mapping(self): + """UDINT should map to UInt32 (unsigned 32-bit integer).""" + assert map_plc_to_opcua_type("UDINT") == ua.VariantType.UInt32 + assert map_plc_to_opcua_type("udint") == ua.VariantType.UInt32 + + def test_dword_mapping(self): + """DWORD should map to UInt32.""" + assert map_plc_to_opcua_type("DWORD") == ua.VariantType.UInt32 + assert map_plc_to_opcua_type("dword") == ua.VariantType.UInt32 + def test_lint_mapping(self): """LINT should map to Int64.""" assert map_plc_to_opcua_type("LINT") == ua.VariantType.Int64 assert map_plc_to_opcua_type("lint") == ua.VariantType.Int64 + def test_ulint_mapping(self): + """ULINT should map to UInt64 (unsigned 64-bit integer).""" + assert map_plc_to_opcua_type("ULINT") == ua.VariantType.UInt64 + assert map_plc_to_opcua_type("ulint") == ua.VariantType.UInt64 + + def test_lword_mapping(self): + """LWORD should map to UInt64.""" + assert map_plc_to_opcua_type("LWORD") == ua.VariantType.UInt64 + assert map_plc_to_opcua_type("lword") == ua.VariantType.UInt64 + def test_float_mapping(self): """FLOAT should map to Float.""" assert map_plc_to_opcua_type("FLOAT") == ua.VariantType.Float @@ -73,6 +113,11 @@ def test_real_mapping(self): assert map_plc_to_opcua_type("REAL") == ua.VariantType.Float assert map_plc_to_opcua_type("real") == ua.VariantType.Float + def test_lreal_mapping(self): + """LREAL should map to Double (IEC 61131-3 LREAL = 64-bit float).""" + assert map_plc_to_opcua_type("LREAL") == ua.VariantType.Double + assert map_plc_to_opcua_type("lreal") == ua.VariantType.Double + def test_string_mapping(self): """STRING should map to String.""" assert map_plc_to_opcua_type("STRING") == ua.VariantType.String @@ -121,6 +166,23 @@ def test_bool_from_false(self): assert convert_value_for_opcua("BOOL", False) is False assert convert_value_for_opcua("BOOL", 0) is False + # SINT conversions (signed 8-bit integer) + def test_sint_normal_values(self): + """Normal SINT values should pass through.""" + assert convert_value_for_opcua("SINT", 0) == 0 + assert convert_value_for_opcua("SINT", 50) == 50 + assert convert_value_for_opcua("SINT", -50) == -50 + + def test_sint_boundary_values(self): + """SINT boundary values should be preserved.""" + assert convert_value_for_opcua("SINT", 127) == 127 + assert convert_value_for_opcua("SINT", -128) == -128 + + def test_sint_clamping(self): + """SINT values outside range should be clamped.""" + assert convert_value_for_opcua("SINT", 200) == 127 + assert convert_value_for_opcua("SINT", -200) == -128 + # BYTE conversions def test_byte_normal_values(self): """Normal byte values should pass through.""" @@ -438,6 +500,13 @@ def test_bool_roundtrip(self): plc_val = convert_value_for_plc("BOOL", opcua_val) assert plc_val == int(val) + def test_sint_roundtrip(self): + """SINT values should survive round-trip conversion.""" + for val in [0, 1, -1, 50, -50, 127, -128]: + opcua_val = convert_value_for_opcua("SINT", val) + plc_val = convert_value_for_plc("SINT", opcua_val) + assert plc_val == val + def test_byte_roundtrip(self): """BYTE values should survive round-trip conversion.""" for val in [0, 1, 127, 128, 255]: From f86fec78afaa331acfef2107df209cb861c55ca2 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Jan 2026 00:56:50 +0000 Subject: [PATCH 2/2] refactor: Clean up type checking in OPC-UA plugin - Remove redundant mixed-case strings from datatype checks since .upper() is already called (e.g., "Sint" is unreachable after "sint".upper() == "SINT") - Replace bare except clauses with specific struct.error catches to avoid catching unintended exceptions like KeyboardInterrupt https://claude.ai/code/session_01FULyWikeb5YVJX8NVX5hax --- .../plugins/python/opcua/opcua_utils.py | 52 +++++++++---------- 1 file changed, 26 insertions(+), 26 deletions(-) diff --git a/core/src/drivers/plugins/python/opcua/opcua_utils.py b/core/src/drivers/plugins/python/opcua/opcua_utils.py index 41748a56..8b283e0e 100644 --- a/core/src/drivers/plugins/python/opcua/opcua_utils.py +++ b/core/src/drivers/plugins/python/opcua/opcua_utils.py @@ -94,7 +94,7 @@ def convert_value_for_opcua(datatype: str, value: Any) -> Any: # The debug utils return raw integer values based on variable size # Convert to appropriate OPC-UA types based on config datatype try: - if datatype.upper() in ["BOOL", "Bool"]: + if datatype.upper() == "BOOL": # Ensure BOOL values are proper Python booleans if isinstance(value, bool): return value @@ -103,39 +103,39 @@ def convert_value_for_opcua(datatype: str, value: Any) -> Any: else: return bool(value) - elif datatype.upper() in ["SINT", "Sint"]: + elif datatype.upper() == "SINT": # Ensure proper int8 type for OPC-UA compatibility (signed 8-bit) clamped_value = max(-128, min(127, int(value))) return ctypes.c_int8(clamped_value).value - elif datatype.upper() in ["BYTE", "Byte", "USINT", "Usint"]: + elif datatype.upper() in ["BYTE", "USINT"]: # Ensure proper uint8 type for OPC-UA compatibility return ctypes.c_uint8(max(0, min(255, int(value)))).value - elif datatype.upper() in ["UINT", "Uint", "WORD", "Word"]: + elif datatype.upper() in ["UINT", "WORD"]: # Ensure proper uint16 type for OPC-UA compatibility clamped_value = max(0, min(65535, int(value))) return ctypes.c_uint16(clamped_value).value - elif datatype.upper() in ["INT", "Int"]: + elif datatype.upper() == "INT": # Ensure proper int16 type - critical for OPC-UA compatibility clamped_value = max(-32768, min(32767, int(value))) return ctypes.c_int16(clamped_value).value - elif datatype.upper() in ["DINT", "Dint", "INT32", "Int32"]: + elif datatype.upper() in ["DINT", "INT32"]: # Ensure proper int32 type for OPC-UA compatibility clamped_value = max(-2147483648, min(2147483647, int(value))) return ctypes.c_int32(clamped_value).value - elif datatype.upper() in ["UDINT", "Udint", "DWORD", "Dword"]: + elif datatype.upper() in ["UDINT", "DWORD"]: # Ensure proper uint32 type for OPC-UA compatibility clamped_value = max(0, min(4294967295, int(value))) return ctypes.c_uint32(clamped_value).value - elif datatype.upper() in ["LINT", "Lint"]: + elif datatype.upper() == "LINT": return int(value) # int64 - elif datatype.upper() in ["ULINT", "Ulint", "LWORD", "Lword"]: + elif datatype.upper() in ["ULINT", "LWORD"]: # Ensure proper uint64 type for OPC-UA compatibility clamped_value = max(0, min(18446744073709551615, int(value))) return ctypes.c_uint64(clamped_value).value @@ -146,21 +146,21 @@ def convert_value_for_opcua(datatype: str, value: Any) -> Any: if isinstance(value, int): try: return struct.unpack('f', struct.pack('I', value))[0] - except: + except struct.error: return float(value) return float(value) - elif datatype.upper() in ["LREAL", "Lreal"]: + elif datatype.upper() == "LREAL": # LREAL (64-bit float) values are stored as integers in debug variables # Convert back to double if it's an integer representation if isinstance(value, int): try: return struct.unpack('d', struct.pack('Q', value))[0] - except: + except struct.error: return float(value) return float(value) - elif datatype.upper() in ["STRING", "String"]: + elif datatype.upper() == "STRING": return str(value) elif datatype.upper() == "TIME": @@ -264,7 +264,7 @@ def convert_value_for_plc(datatype: str, value: Any) -> Any: """Convert OPC-UA value to PLC debug variable format.""" # Handle different OPC-UA value types more robustly try: - if datatype.upper() in ["BOOL", "Bool"]: + if datatype.upper() == "BOOL": # Convert any value to boolean, then to int (0/1) if isinstance(value, bool): return int(value) @@ -275,39 +275,39 @@ def convert_value_for_plc(datatype: str, value: Any) -> Any: else: return int(bool(value)) - elif datatype.upper() in ["SINT", "Sint"]: + elif datatype.upper() == "SINT": # Ensure proper int8 type for PLC compatibility (signed 8-bit) clamped_value = max(-128, min(127, int(value))) return ctypes.c_int8(clamped_value).value - elif datatype.upper() in ["BYTE", "Byte", "USINT", "Usint"]: + elif datatype.upper() in ["BYTE", "USINT"]: # Ensure proper uint8 type for PLC compatibility return ctypes.c_uint8(max(0, min(255, int(value)))).value - elif datatype.upper() in ["INT", "Int"]: + elif datatype.upper() == "INT": # Ensure proper int16 type for PLC compatibility clamped_value = max(-32768, min(32767, int(value))) return ctypes.c_int16(clamped_value).value - elif datatype.upper() in ["UINT", "Uint", "WORD", "Word"]: + elif datatype.upper() in ["UINT", "WORD"]: # Ensure proper uint16 type for PLC compatibility clamped_value = max(0, min(65535, int(value))) return ctypes.c_uint16(clamped_value).value - elif datatype.upper() in ["DINT", "Dint", "INT32", "Int32"]: + elif datatype.upper() in ["DINT", "INT32"]: # Ensure proper int32 type for PLC compatibility clamped_value = max(-2147483648, min(2147483647, int(value))) return ctypes.c_int32(clamped_value).value - elif datatype.upper() in ["UDINT", "Udint", "DWORD", "Dword"]: + elif datatype.upper() in ["UDINT", "DWORD"]: # Ensure proper uint32 type for PLC compatibility clamped_value = max(0, min(4294967295, int(value))) return ctypes.c_uint32(clamped_value).value - elif datatype.upper() in ["LINT", "Lint"]: + elif datatype.upper() == "LINT": return int(value) # int64 - elif datatype.upper() in ["ULINT", "Ulint", "LWORD", "Lword"]: + elif datatype.upper() in ["ULINT", "LWORD"]: # Ensure proper uint64 type for PLC compatibility clamped_value = max(0, min(18446744073709551615, int(value))) return ctypes.c_uint64(clamped_value).value @@ -317,22 +317,22 @@ def convert_value_for_plc(datatype: str, value: Any) -> Any: if isinstance(value, float): try: return struct.unpack('I', struct.pack('f', value))[0] - except: + except struct.error: return int(value) else: return int(float(value)) - elif datatype.upper() in ["LREAL", "Lreal"]: + elif datatype.upper() == "LREAL": # Convert double to int representation for storage (64-bit) if isinstance(value, float): try: return struct.unpack('Q', struct.pack('d', value))[0] - except: + except struct.error: return int(value) else: return int(float(value)) - elif datatype.upper() in ["STRING", "String"]: + elif datatype.upper() == "STRING": return str(value) elif datatype.upper() == "TIME":