Fix OPC-UA type mappings

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)
@@ -79,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
@@ -88,34 +103,64 @@ def convert_value_for_opcua(datatype: str, value: Any) -> Any:
             else:
                 return bool(value)
 
-        elif datatype.upper() in ["BYTE", "Byte"]:
+        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", "USINT"]:
             # Ensure proper uint8 type for OPC-UA compatibility
             return ctypes.c_uint8(max(0, min(255, int(value)))).value
-
-        elif datatype.upper() in ["INT", "Int"]:
+
+        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() == "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 ["LINT", "Lint"]:
+
+        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() == "LINT":
             return int(value)  # int64
-
+
+        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
+
         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
             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 ["STRING", "String"]:
+
+        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 struct.error:
+                    return float(value)
+            return float(value)
+
+        elif datatype.upper() == "STRING":
             return str(value)
 
         elif datatype.upper() == "TIME":
@@ -219,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)
@@ -230,34 +275,64 @@ def convert_value_for_plc(datatype: str, value: Any) -> Any:
             else:
                 return int(bool(value))
 
-        elif datatype.upper() in ["BYTE", "Byte"]:
+        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", "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  
+
+        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 ["DINT", "Dint", "INT32", "Int32"]:
+
+        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", "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 ["LINT", "Lint"]:
+
+        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() == "LINT":
             return int(value)  # int64
-
+
+        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
+
         elif datatype.upper() in ["FLOAT", "REAL"]:
             # Convert float to int representation for storage
             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 ["STRING", "String"]:
+
+        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 struct.error:
+                    return int(value)
+            else:
+                return int(float(value))
+
+        elif datatype.upper() == "STRING":
             return str(value)
 
         elif datatype.upper() == "TIME":

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]: