Added fallback to type detection to allow automatic ENUM usage + Added exception raising for unknown types

Author: RobertoRoos

pyads/connection.py

@@ -91,6 +91,7 @@
     _dict_slice_generator,
     bytes_from_dict,
     size_of_structure,
+    ADSError,
 )
 from .symbol import AdsSymbol
 from .utils import decode_ads
@@ -173,16 +174,16 @@ def _query_plc_datatype_from_name(self, data_name: str,
 
         If cache_symbol_info is True then the SymbolInfo will be cached and adsGetSymbolInfo
         will only used once.
-
         """
+        info = None
         if cache_symbol_info:
             info = self._symbol_info_cache.get(data_name)
-            if info is None:
-                info = adsGetSymbolInfo(self._port, self._adr, data_name)
-                self._symbol_info_cache[data_name] = info
-        else:
+        if info is None:
             info = adsGetSymbolInfo(self._port, self._adr, data_name)
-        return AdsSymbol.get_type_from_str(info.symbol_type)
+        if cache_symbol_info:
+            self._symbol_info_cache[data_name] = info
+
+        return AdsSymbol.get_type_from_info(info)
 
     def open(self) -> None:
         """Connect to the TwinCAT message router."""
@@ -537,6 +538,10 @@ def read_by_name(
             plc_datatype = self._query_plc_datatype_from_name(data_name,
                                                               cache_symbol_info)
 
+        if plc_datatype is None:
+            # `adsSyncReadReqEx2()` will fail for a None type
+            raise ADSError(None, f"Failed to detect datatype for `{data_name}`")
+
         return adsSyncReadByNameEx(
             self._port,
             self._adr,
@@ -682,6 +687,10 @@ def write_by_name(
             plc_datatype = self._query_plc_datatype_from_name(data_name,
                                                               cache_symbol_info)
 
+        if plc_datatype is None:
+            # `adsSyncWriteReqEx()` will fail for a None type
+            raise ADSError(None, f"Failed to detect datatype for `{data_name}`")
+
         return adsSyncWriteByNameEx(
             self._port, self._adr, data_name, value, plc_datatype, handle=handle
         )

pyads/symbol.py

@@ -16,9 +16,10 @@
 from typing import TYPE_CHECKING, Any, Optional, List, Tuple, Callable, Union, Type
 
 from . import constants  # To access all constants, use package notation
-from .constants import PLCDataType
+from .constants import PLCDataType, ads_type_to_ctype
 from .pyads_ex import adsGetSymbolInfo
-from .structs import NotificationAttrib
+from .structs import NotificationAttrib, SAdsSymbolEntry
+from .ads import ADSError
 
 # ads.Connection relies on structs.AdsSymbol (but in type hints only), so use
 # this 'if' to only include it when type hinting (False during execution)
@@ -125,7 +126,7 @@ def __init__(
         self.name = name
         self.index_offset = index_offset
         self.index_group = index_group
-        self.symbol_type = symbol_type
+        self.symbol_type = None  # Apply `symbol_type` later
         self.comment = comment
         self._value: Any = None
 
@@ -137,19 +138,18 @@ def __init__(
             from .ads import size_of_structure
             self._structure_size = size_of_structure(self.structure_def * self.array_size)
 
+        self.plc_type: Optional[Type[PLCDataType]] = None
+
         if missing_info:
             self._create_symbol_from_info()  # Perform remote lookup
 
-        # Now `self.symbol_type` should have a value, find the actual PLCTYPE
-        # from it.
-        # This is relevant for both lookup and full user definition.
-
-        self.plc_type: Optional[Type[PLCDataType]] = None
-        if self.symbol_type is not None:
+        # Apply user-provided type (overriding auto detect if any):
+        if symbol_type is not None:
             if isinstance(self.symbol_type, str):  # Perform lookup if string
                 self.plc_type = AdsSymbol.get_type_from_str(self.symbol_type)
             else:  # Otherwise `symbol_type` is probably a pyads.PLCTYPE_* constant
                 self.plc_type = self.symbol_type
+            self.symbol_type = symbol_type
 
         self.auto_update = auto_update
 
@@ -166,12 +166,8 @@ def _create_symbol_from_info(self) -> None:
         if info.comment:
             self.comment = info.comment
 
-        # info.dataType is an integer mapping to a type in
-        # constants.ads_type_to_ctype.
-        # However, this type ignores whether the variable is really an array!
-        # So are not going to be using this and instead rely on the textual
-        # type
-        self.symbol_type = info.symbol_type  # Save the type as string
+        self.plc_type = self.get_type_from_info(info)
+        self.symbol_type = info.symbol_type  # Also save the type as string
 
     def _check_for_open_connection(self) -> None:
         """Assert the current object is ready to read from/write to.
@@ -195,6 +191,12 @@ def read(self) -> Any:
                                                            structure_size=self._structure_size,
                                                            array_size=self.array_size)
         else:
+            if self.plc_type is None:
+                raise ADSError(
+                    None,
+                    f"Cannot read data with unknown datatype for symbol "
+                    f"{self.name} ({self.symbol_type})"
+                )
             self._value = self._plc.read(self.index_group, self.index_offset, self.plc_type)
 
         return self._value
@@ -218,6 +220,12 @@ def write(self, new_value: Optional[Any] = None) -> None:
             self._plc.write_structure_by_name(self.name, new_value, self.structure_def,
                                               structure_size=self._structure_size, array_size=self.array_size)
         else:
+            if self.plc_type is None:
+                raise ADSError(
+                    None,
+                    f"Cannot write data with unknown datatype for symbol "
+                    f"{self.name} ({self.symbol_type})"
+                )
             self._plc.write(self.index_group, self.index_offset, new_value, self.plc_type)
 
     def __repr__(self) -> str:
@@ -283,6 +291,30 @@ def _value_callback(self, notification: Any, data_name: Any) -> None:
         )
         self._value = value
 
+    @classmethod
+    def get_type_from_info(cls, info: SAdsSymbolEntry) -> Optional[Type[PLCDataType]]:
+        """Get PLCTYPE_* from symbol info struct
+
+        Also see :meth:`~get_type_from_str`.
+        """
+        plc_type = cls.get_type_from_str(info.symbol_type)
+        if plc_type is not None:
+            return plc_type
+
+        # Failed to detect by name (e.g. type is enum)
+
+        # Use `ADST_*` integer to detect type instead
+        plc_type = ads_type_to_ctype.get(info.dataType, None)
+        if plc_type is not None:
+            array_size = int(info.size / sizeof(plc_type))
+            # However, `dataType` is always a scalar, even if the object is an array:
+            if array_size > 1:
+                plc_type = plc_type * array_size
+
+            return plc_type
+
+        return None
+
     @staticmethod
     def get_type_from_str(type_str: str) -> Optional[Type[PLCDataType]]:
         """Get PLCTYPE_* from PLC name string