"""ADS Connection class.
:author: Stefan Lehmann <stlm@posteo.de>
:maintainer: Filippo Boido <filippo.boido@agileautomation.eu> (Agile Automation Technologies GmbH)
:license: MIT, see license file or https://opensource.org/licenses/MIT
:created on: 2018-06-11 18:15:53
"""
from __future__ import annotations
import inspect
import struct
from ctypes import (
memmove,
addressof,
c_ubyte,
Array,
Structure,
sizeof,
)
from datetime import datetime
from typing import Optional, Union, Tuple, Any, Type, Callable, Dict, List, Mapping, Sequence, cast, TypeVar, overload, Literal
# noinspection PyUnresolvedReferences
from .constants import (
ADSIGRP_SYM_UPLOAD,
ADSIGRP_SYM_DT_UPLOAD,
ADSIGRP_SYM_INFOBYNAMEEX,
ADSIGRP_SYM_UPLOADINFO2,
ADSIGRP_SYM_VERSION,
ADSIOFFS_DEVDATA_ADSSTATE,
PLCTYPE_BOOL,
PLCTYPE_BYTE,
PLCTYPE_DATE,
PLCTYPE_DINT,
PLCTYPE_DT,
PLCTYPE_DWORD,
PLCTYPE_INT,
PLCTYPE_LREAL,
PLCTYPE_REAL,
PLCTYPE_SINT,
PLCTYPE_STRING,
PLCTYPE_TIME,
PLCTYPE_TOD,
PLCTYPE_UDINT,
PLCTYPE_UINT,
PLCTYPE_USINT,
PLCTYPE_WORD,
PLC_DEFAULT_STRING_SIZE,
DATATYPE_MAP,
ADSIGRP_SUMUP_READ,
ADSIGRP_SUMUP_WRITE,
ADSIGRP_SYM_VALBYHND,
MAX_ADS_SUB_COMMANDS,
ads_type_to_ctype,
PLCSimpleDataType,
PLCDataType,
)
from .filetimes import filetime_to_dt
from .pyads_ex import (
adsAddRoute,
adsDelRoute,
adsPortOpenEx,
adsPortCloseEx,
adsGetLocalAddressEx,
adsSyncReadStateReqEx,
adsSyncReadDeviceInfoReqEx,
adsSyncWriteControlReqEx,
adsSyncWriteReqEx,
adsSyncReadWriteReqEx2,
adsSyncReadReqEx2,
adsGetHandle,
adsGetNetIdForPLC,
adsGetSymbolInfo,
adsSumRead,
adsSumReadDetailed,
adsSumWrite,
adsSumWriteDetailed,
_sum_symbol_unsupported_reason,
adsReleaseHandle,
adsSyncReadByNameEx,
adsSyncWriteByNameEx,
adsSyncAddDeviceNotificationReqEx,
adsSyncDelDeviceNotificationReqEx,
adsSyncSetTimeoutEx,
_clear_notification_state_for_address,
type_is_string,
type_is_wstring,
ADSError,
)
from .errorcodes import ERROR_CODES
from .structs import (
AmsAddr,
AdsVersion,
NotificationAttrib,
SAdsNotificationHeader,
SAdsSymbolEntry,
)
from .ads import (
linux,
StructureDef,
dict_from_bytes,
_list_slice_generator,
_dict_slice_generator,
bytes_from_dict,
size_of_structure,
)
from .symbol import AdsSymbol
from .rpc_interface import resolve_rpc_interface_definition
from .metadata import (
AdsDataType,
AdsSymbolInfo,
AdsUploadInfo,
SumReadResult,
SumWriteResult,
parse_data_type_entries,
parse_symbol_entries,
parse_symbol_entry,
parse_upload_info,
)
RpcInterfaceT = TypeVar("RpcInterfaceT")
def _validated_exact_symbol_metadata(
requested_names: Sequence[str],
symbol_metadata: Mapping[str, AdsSymbolInfo],
) -> Dict[str, AdsSymbolInfo]:
"""Snapshot and validate approved immutable symbol layouts before I/O."""
if not isinstance(symbol_metadata, Mapping):
raise TypeError("symbol_metadata must be a mapping of names to AdsSymbolInfo")
supplied = dict(symbol_metadata.items())
requested = tuple(requested_names)
for name in requested:
if not isinstance(name, str) or not name or "\x00" in name:
raise ValueError("ADS symbol names must be non-empty strings without NUL")
requested_keys = set(requested)
supplied_keys = set(supplied)
if supplied_keys != requested_keys:
missing = sorted(repr(name) for name in requested_keys - supplied_keys)
unexpected = sorted(repr(name) for name in supplied_keys - requested_keys)
raise ValueError(
"symbol_metadata keys must exactly match requested symbol names "
f"(missing=[{', '.join(missing)}], "
f"unexpected=[{', '.join(unexpected)}])"
)
validated: Dict[str, AdsSymbolInfo] = {}
for name in requested:
info = supplied[name]
if not isinstance(info, AdsSymbolInfo):
raise TypeError(
f"symbol_metadata[{name!r}] must be an immutable AdsSymbolInfo"
)
if not isinstance(info.name, str) or info.name != name:
raise ValueError(
f"symbol_metadata[{name!r}].name must exactly equal the mapping key"
)
if "\x00" in info.name:
raise ValueError(f"symbol_metadata[{name!r}].name contains NUL")
if not isinstance(info.type_name, str):
raise TypeError(f"symbol_metadata[{name!r}].type_name must be str")
if "\x00" in info.type_name:
raise ValueError(f"symbol_metadata[{name!r}].type_name contains NUL")
for field_name in (
"index_group",
"index_offset",
"size",
"ads_data_type",
"flags",
):
value = getattr(info, field_name)
if type(value) is not int:
raise TypeError(
f"symbol_metadata[{name!r}].{field_name} must be int"
)
if not 0 <= value <= 0xFFFFFFFF:
raise ValueError(
f"symbol_metadata[{name!r}].{field_name} must fit uint32"
)
validated[name] = info
return validated
[docs]
class RpcObject:
"""Proxy object for calling PLC RPC methods using native attribute syntax."""
def __init__(
self,
connection: "Connection",
object_name: str,
method_separator: str = "#",
method_prefixes: Tuple[str, ...] = ("", "m_"),
method_return_types: Optional[Dict[str, Type["PLCDataType"]]] = None,
method_parameters: Optional[
Dict[str, Sequence[Type["PLCDataType"]]]
] = None,
) -> None:
self._connection = connection
self._object_name = object_name
self._method_separator = method_separator
self._method_prefixes = method_prefixes
self._method_return_types: Dict[str, Type["PLCDataType"]] = (
method_return_types.copy() if method_return_types else {}
)
self._method_parameters: Dict[str, Tuple[Type["PLCDataType"], ...]] = (
{name: tuple(values) for name, values in method_parameters.items()}
if method_parameters
else {}
)
# If a method is declared in `method_return_types` but no explicit
# parameter signature is provided, treat it as a zero-argument RPC.
for method_name in self._method_return_types:
self._method_parameters.setdefault(method_name, tuple())
self._return_spec_cache: Dict[
str, Tuple[Optional[Type["PLCDataType"]], bool]
] = {}
def _candidate_method_names(self, method_name: str) -> List[str]:
names: List[str] = []
for prefix in self._method_prefixes:
if method_name.startswith(prefix):
candidate = method_name
else:
candidate = f"{prefix}{method_name}"
names.append(f"{self._object_name}{self._method_separator}{candidate}")
# Keep order but drop duplicates
return list(dict.fromkeys(names))
def _get_symbol_info(self, qualified_name: str) -> Optional[SAdsSymbolEntry]:
"""Return cached symbol info for a fully qualified RPC method name."""
if not self._connection.is_open or self._connection._port is None:
return None
cache = self._connection._symbol_info_cache
info = cache.get(qualified_name)
if info is not None:
return info
try:
info = adsGetSymbolInfo(
self._connection._port, self._connection._adr, qualified_name
)
except ADSError:
return None
cache[qualified_name] = info
return info
[docs]
def set_return_type(self, method_name: str, plc_type: Type["PLCDataType"]) -> None:
"""Register or override the expected return type for a method."""
self._method_return_types[method_name] = plc_type
self._return_spec_cache.pop(
f"{self._object_name}{self._method_separator}{method_name}", None
)
def _manual_return_type(self, method_name: str) -> Optional[Type["PLCDataType"]]:
"""Look up manually configured return type for a method name."""
candidates = {method_name}
for prefix in self._method_prefixes:
if method_name.startswith(prefix):
candidates.add(method_name[len(prefix):])
else:
candidates.add(f"{prefix}{method_name}")
for candidate in candidates:
plc_type = self._method_return_types.get(candidate)
if plc_type is not None:
return plc_type
return None
def _infer_return_spec(
self, qualified_name: str
) -> Tuple[Optional[Type["PLCDataType"]], bool]:
"""Infer the PLC return type for a method, falling back to raw bytes."""
cached = self._return_spec_cache.get(qualified_name)
if cached is not None:
return cached
plc_type: Optional[Type["PLCDataType"]] = None
coerce_bytes = False
_, _, method_name = qualified_name.partition(self._method_separator)
if method_name:
plc_type = self._manual_return_type(method_name)
if plc_type is None:
info = self._get_symbol_info(qualified_name)
if info is not None:
plc_type = ads_type_to_ctype.get(info.dataType)
if plc_type is None:
plc_type = AdsSymbol.get_type_from_str(info.symbol_type)
if plc_type is None and info.size:
plc_type = c_ubyte * info.size
coerce_bytes = True
spec = (plc_type, coerce_bytes)
self._return_spec_cache[qualified_name] = spec
return spec
@staticmethod
def _normalize_control_args(
args: Tuple[Any, ...], kwargs: Dict[str, Any]
) -> Tuple[Any, Optional[Type["PLCDataType"]], Optional[Type["PLCDataType"]], Optional[str]]:
write_value = kwargs.pop("write_value", None)
write_type = kwargs.pop("write_type", None)
return_type = kwargs.pop("return_type", None)
method_name_override = kwargs.pop("method_name", None)
if kwargs:
unknown = ", ".join(sorted(kwargs.keys()))
raise TypeError(f"Unknown keyword argument(s): {unknown}")
if len(args) > 3:
raise TypeError("Expected up to 3 positional args: value, write_type, return_type")
if len(args) >= 1:
if write_value is not None:
raise TypeError("write_value provided both positionally and as keyword")
write_value = args[0]
if len(args) >= 2:
if write_type is not None:
raise TypeError("write_type provided both positionally and as keyword")
write_type = args[1]
if len(args) == 3:
if return_type is not None:
raise TypeError("return_type provided both positionally and as keyword")
return_type = args[2]
return write_value, write_type, return_type, method_name_override
def _parameter_types_for(self, method_name: str) -> Tuple[Type["PLCDataType"], ...]:
candidates = {method_name}
for prefix in self._method_prefixes:
if method_name.startswith(prefix):
candidates.add(method_name[len(prefix):])
else:
candidates.add(f"{prefix}{method_name}")
for candidate in candidates:
values = self._method_parameters.get(candidate)
if values is not None:
return values
return tuple()
@staticmethod
def _pack_single_argument(value: Any, plc_type: Type["PLCDataType"]) -> bytes:
if type_is_string(plc_type):
return value.encode("utf-8") + b"\x00"
if type_is_wstring(plc_type):
return value.encode("utf-16-le")
if type(plc_type).__name__ == "PyCArrayType":
data = plc_type(*value)
elif type(value) is plc_type:
data = value
else:
data = plc_type(value)
raw = (c_ubyte * sizeof(data))()
memmove(raw, addressof(data), sizeof(data))
return bytes(raw)
def _pack_method_args(
self,
method_name: str,
args: Tuple[Any, ...],
) -> Tuple[Any, Optional[Type["PLCDataType"]]]:
parameter_types = self._parameter_types_for(method_name)
if not parameter_types:
if len(args) == 0:
return None, None
if len(args) == 1:
raise TypeError(
"Parameter types are not configured for this method. "
"Pass `method_parameters=` to get_object, "
"or call with explicit write_type."
)
raise TypeError(
"Parameter types are not configured for this method. "
"Use explicit write_type or configure method parameters."
)
if len(args) != len(parameter_types):
raise TypeError(
f"{method_name} expects {len(parameter_types)} parameter(s), "
f"got {len(args)}."
)
payload = bytearray()
for value, plc_type in zip(args, parameter_types):
payload.extend(self._pack_single_argument(value, plc_type))
if not payload:
return None, None
write_type = c_ubyte * len(payload)
return list(payload), write_type
def __getattr__(self, method_name: str) -> Callable[..., Any]:
if method_name.startswith("_"):
raise AttributeError(method_name)
def _invoke(*args: Any, **kwargs: Any) -> Any:
control_keys = {"write_value", "write_type", "return_type", "method_name"}
has_control_args = any(key in kwargs for key in control_keys)
if has_control_args:
write_value, write_type, return_type, method_name_override = self._normalize_control_args(args, kwargs)
else:
if kwargs:
unknown = ", ".join(sorted(kwargs.keys()))
raise TypeError(f"Unknown keyword argument(s): {unknown}")
is_legacy_positional = (
len(args) <= 3
and (
len(args) == 0
or (len(args) >= 2 and isinstance(args[1], type))
)
and not self._parameter_types_for(method_name)
)
if is_legacy_positional:
write_value = args[0] if len(args) >= 1 else None
write_type = args[1] if len(args) >= 2 else None
return_type = args[2] if len(args) == 3 else None
method_name_override = None
else:
write_value, write_type = self._pack_method_args(method_name, args)
return_type = None
method_name_override = None
if method_name_override is not None:
return self._connection.call_rpc_method(
method_name_override,
return_type=return_type,
write_value=write_value,
write_type=write_type,
)
last_error: Optional[ADSError] = None
for candidate_name in self._candidate_method_names(method_name):
resolved_return_type = return_type
coerce_bytes = False
if resolved_return_type is None:
resolved_return_type, coerce_bytes = self._infer_return_spec(
candidate_name
)
try:
result = self._connection.call_rpc_method(
candidate_name,
return_type=resolved_return_type,
write_value=write_value,
write_type=write_type,
)
except ADSError as exc:
if exc.err_code != 1808: # symbol not found
raise
last_error = exc
continue
if coerce_bytes and isinstance(result, list):
return bytes(result)
return result
if last_error is not None:
raise last_error
raise AttributeError(method_name)
return _invoke
[docs]
class Connection(object):
"""Class for managing the connection to an ADS device.
:ivar str ams_net_id: AMS net id of the remote device
:ivar int ams_net_port: port of the remote device
:ivar str ip_address: the ip address of the device
:note: If no IP address is given the ip address is automatically set
to first 4 parts of the Ams net id.
"""
def __init__(
self, ams_net_id: str = None, ams_net_port: int = None,
ip_address: str = None,
route_policy: Literal["auto", "existing_only"] = "auto",
) -> None:
# Establish destructor-safe state before validating caller input. A
# partially initialized object can otherwise reach ``__del__`` after
# validation raises and mask the useful constructor error.
self._port = None # type: Optional[int]
self._open = False
self._managed_route = False
self._notifications = {} # type: Dict[int, str]
self._symbol_info_cache: Dict[str, SAdsSymbolEntry] = {}
self._upload_info_cache: Optional[AdsUploadInfo] = None
if route_policy not in ("auto", "existing_only"):
raise ValueError("route_policy must be 'auto' or 'existing_only'")
self._adr = AmsAddr(ams_net_id, ams_net_port)
if ip_address is None:
if ams_net_id is None:
raise TypeError("Must provide an IP or net ID")
self.ip_address = ".".join(ams_net_id.split(".")[:4])
else:
self.ip_address = ip_address
self.ams_net_id = ams_net_id
self.ams_net_port = ams_net_port
self.route_policy = route_policy
@property
def ams_netid(self) -> str:
return self._adr.netid
@ams_netid.setter
def ams_netid(self, value: str) -> None:
if self._open:
raise AttributeError(
"Setting netid is not allowed while connection is open."
)
self._adr.netid = value
@property
def ams_port(self) -> int:
return self._adr.port
@ams_port.setter
def ams_port(self, value: int) -> None:
if self._open:
raise AttributeError(
"Setting port is not allowed while connection is open."
)
self._adr.port = value
def __enter__(self) -> "Connection":
"""Open on entering with-block."""
self.open()
return self
def __exit__(self, _type: Type, _val: Any, _traceback: Any) -> None:
"""Close on leaving with-block."""
self.close()
def __del__(self) -> None:
"""Class destructor.
Make sure to close the connection when an instance runs out of scope.
"""
# If the connection is already closed, nothing new will happen
self.close()
def _query_plc_datatype_from_name(self, data_name: str,
cache_symbol_info: bool) -> Type:
"""Return the plc_datatype by reading SymbolInfo from the target.
If cache_symbol_info is True then the SymbolInfo will be cached and adsGetSymbolInfo
will only used once.
"""
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:
info = adsGetSymbolInfo(self._port, self._adr, data_name)
return AdsSymbol.get_type_from_str(info.symbol_type)
[docs]
def open(self) -> None:
"""Connect to the TwinCAT message router."""
if self._open:
return
if self.ams_net_id is None:
self.ams_net_id = adsGetNetIdForPLC(self.ip_address)
self._adr = AmsAddr(self.ams_net_id, self.ams_net_port)
self._port = adsPortOpenEx()
if linux and self.route_policy == "auto":
try:
adsAddRoute(self._adr.netIdStruct(), self.ip_address)
self._managed_route = True
except ADSError:
adsPortCloseEx(self._port)
self._port = None
raise
self._open = True
[docs]
def close(self) -> None:
""":summary: Close the connection to the TwinCAT message router."""
try:
if self._open and linux and self._managed_route:
adsDelRoute(self._adr.netIdStruct())
finally:
self._managed_route = False
try:
if self._port is not None:
adsPortCloseEx(self._port)
finally:
self._port = None
self._open = False
address = getattr(self, "_adr", None)
if address is not None:
_clear_notification_state_for_address(address)
# Symbol handles and layouts are scoped to a runtime
# generation. Never carry inferred metadata across reopen.
self._symbol_info_cache.clear()
self._upload_info_cache = None
[docs]
def clear_symbol_cache(self) -> None:
"""Invalidate runtime-scoped symbol layouts and inferred PLC types."""
self._symbol_info_cache.clear()
self._upload_info_cache = None
def _runtime_string_encoding(self) -> str:
"""Use discovered v3 STRING context without forcing extra ADS calls."""
if self._upload_info_cache is None:
return "utf-8"
return self._upload_info_cache.string_encoding
[docs]
def get_local_address(self) -> Optional[AmsAddr]:
"""Return the local AMS-address and the port number.
:rtype: AmsAddr
"""
if self._port is not None:
return adsGetLocalAddressEx(self._port)
return None
[docs]
def read_state(self) -> Optional[Tuple[int, int]]:
"""Read the current ADS-state and the machine-state.
Read the current ADS-state and the machine-state from the ADS-server.
:rtype: (int, int)
:return: adsState, deviceState
"""
if self._port is not None:
return adsSyncReadStateReqEx(self._port, self._adr)
return None
[docs]
def write_control(
self, ads_state: int, device_state: int, data: Any, plc_datatype: Type
) -> None:
"""Change the ADS state and the machine-state of the ADS-server.
:param int ads_state: new ADS-state, according to ADSTATE constants
:param int device_state: new machine-state
:param data: additional data
:param int plc_datatype: datatype, according to PLCTYPE constants
:note: Despite changing the ADS-state and the machine-state it is
possible to send additional data to the ADS-server. For current
ADS-devices additional data is not progressed.
Every ADS-device is able to communicate its current state to other
devices. There is a difference between the device-state and the
state of the ADS-interface (AdsState). The possible states of an
ADS-interface are defined in the ADS-specification.
"""
if self._port is not None:
return adsSyncWriteControlReqEx(
self._port, self._adr, ads_state, device_state, data, plc_datatype
)
[docs]
def read_device_info(self) -> Optional[Tuple[str, AdsVersion]]:
"""Read the name and the version number of the ADS-server.
:rtype: string, AdsVersion
:return: device name, version
"""
if self._port is not None:
return adsSyncReadDeviceInfoReqEx(self._port, self._adr)
return None
[docs]
def write(
self, index_group: int, index_offset: int, value: Any,
plc_datatype: Type["PLCDataType"]
) -> None:
"""Send data synchronous to an ADS-device.
:param int index_group: PLC storage area, according to the INDEXGROUP
constants
:param int index_offset: PLC storage address
:param Any value: value to write to the storage address of the PLC
:param Type["PLCDataType"] plc_datatype: type of the data given to the PLC,
according to PLCTYPE constants
"""
if self._port is not None:
return adsSyncWriteReqEx(
self._port, self._adr, index_group, index_offset, value, plc_datatype
)
[docs]
def read_write(
self,
index_group: int,
index_offset: int,
plc_read_datatype: Optional[Type["PLCDataType"]],
value: Any,
plc_write_datatype: Optional[Type["PLCDataType"]],
return_ctypes: bool = False,
check_length: bool = True,
) -> Any:
"""Read and write data synchronous from/to an ADS-device.
:param int index_group: PLC storage area, according to the INDEXGROUP
constants
:param int index_offset: PLC storage address
:param Type["PLCDataType"] plc_read_datatype: type of the data given to the PLC to respond to,
according to PLCTYPE constants, or None to not read anything
:param value: value to write to the storage address of the PLC
:param Type["PLCDataType"] plc_write_datatype: type of the data given to the PLC, according to
PLCTYPE constants, or None to not write anything
:param bool return_ctypes: return ctypes instead of python types if True (default: False)
:param bool check_length: check whether the amount of bytes read matches the size
of the read data type (default: True)
:return: value: **value**
"""
if self._port is not None:
return adsSyncReadWriteReqEx2(
self._port,
self._adr,
index_group,
index_offset,
plc_read_datatype,
value,
plc_write_datatype,
return_ctypes,
check_length,
)
return None
[docs]
def read(
self,
index_group: int,
index_offset: int,
plc_datatype: Type["PLCDataType"],
return_ctypes: bool = False,
check_length: bool = True,
) -> Any:
"""Read data synchronous from an ADS-device.
:param int index_group: PLC storage area, according to the INDEXGROUP
constants
:param int index_offset: PLC storage address
:param Type["PLCDataType"] plc_datatype: type of the data given to the PLC, according
to PLCTYPE constants
:param bool return_ctypes: return ctypes instead of python types if True
(default: False)
:param bool check_length: check whether the amount of bytes read matches the size
of the read data type (default: True)
:return: value
"""
if index_group is None or not isinstance(index_group, int):
raise TypeError('index_group: integer is required')
if index_offset is None or not isinstance(index_offset, int):
raise TypeError('index_offset: integer is required')
if self._port is not None:
return adsSyncReadReqEx2(
self._port,
self._adr,
index_group,
index_offset,
plc_datatype,
return_ctypes,
check_length,
)
return None
[docs]
def get_symbol(
self,
name: Optional[str] = None,
index_group: Optional[int] = None,
index_offset: Optional[int] = None,
plc_datatype: Optional[Union[Type["PLCDataType"], str]] = None,
comment: Optional[str] = None,
auto_update: bool = False,
structure_def: Optional["StructureDef"] = None,
array_size: Optional[int] = 1,
) -> AdsSymbol:
"""Create a symbol instance
Specify either the variable name or the index_group **and**
index_offset so the symbol can be located.
If the name was specified but not all other attributes were,
the other attributes will be looked up from the connection.
`data_type` can be a PLCTYPE constant or a string representing
a PLC type (e.g. 'LREAL').
:param str name:
:param Optional[int] index_group:
:param Optional[int] index_offset:
:param plc_datatype: type of the PLC variable, according
to PLCTYPE constants
:param str comment: comment
:param bool auto_update: Create notification to update buffer (same as
`set_auto_update(True)`)
:param Optional["StructureDef"] structure_def: special tuple defining the structure and
types contained within it according to PLCTYPE constants, must match
the structure defined in the PLC, PLC structure must be defined with
{attribute 'pack_mode' := '1'}
:param Optional[int] array_size: size of array if reading array of structure, defaults to 1
Expected input example for structure_def:
.. code:: python
structure_def = (
('rVar', pyads.PLCTYPE_LREAL, 1),
('sVar', pyads.PLCTYPE_STRING, 2, 35),
('SVar1', pyads.PLCTYPE_STRING, 1),
('rVar1', pyads.PLCTYPE_REAL, 1),
('iVar', pyads.PLCTYPE_DINT, 1),
('iVar1', pyads.PLCTYPE_INT, 3),
)
# i.e ('Variable Name', variable type, arr size (1 if not array),
# length of string (if defined in PLC))
"""
return AdsSymbol(self, name, index_group, index_offset, plc_datatype,
comment, auto_update=auto_update, structure_def=structure_def,
array_size=array_size)
def _require_open_port(self) -> int:
"""Return the native port or fail instead of hiding a disconnect."""
if self._port is None or not self._open:
raise RuntimeError("Connection must be open for ADS metadata discovery.")
return self._port
[docs]
def get_upload_info(self) -> AdsUploadInfo:
"""Return immutable symbol/datatype upload sizes for this runtime."""
port = self._require_open_port()
upload_type = c_ubyte * 64
payload, bytes_read = adsSyncReadReqEx2(
port,
self._adr,
ADSIGRP_SYM_UPLOADINFO2,
ADSIOFFS_DEVDATA_ADSSTATE,
upload_type,
return_ctypes=True,
check_length=False,
return_bytes_read=True,
)
upload = parse_upload_info(bytes(payload)[:bytes_read])
# Fail explicitly if a modern target advertises a code page Python
# cannot decode; never reinterpret symbol text as cp1252 silently.
upload.symbol_encoding
self._upload_info_cache = upload
return upload
[docs]
def get_symbol_version(self) -> int:
"""Return TwinCAT's authoritative symbol-table generation counter."""
port = self._require_open_port()
return int(
adsSyncReadReqEx2(
port,
self._adr,
ADSIGRP_SYM_VERSION,
0,
PLCTYPE_UDINT,
)
)
[docs]
def get_symbol_info(self, name: str) -> AdsSymbolInfo:
"""Return immutable native metadata for one exact symbol name."""
port = self._require_open_port()
upload = self.get_upload_info()
# Probe the variable-length entry header without asking the legacy
# fixed-buffer helper to enforce the advertised complete length.
native = adsSyncReadWriteReqEx2(
port,
self._adr,
ADSIGRP_SYM_INFOBYNAMEEX,
0,
SAdsSymbolEntry,
name,
PLCTYPE_STRING,
return_ctypes=True,
check_length=False,
)
if native.entryLength <= sizeof(native):
return parse_symbol_entry(
bytes(native)[:native.entryLength],
encoding=upload.symbol_encoding,
)
# SAdsSymbolEntry intentionally retains its historical fixed string
# buffer. Query an exact-size byte array when TwinCAT advertises a
# larger entry so comments/extensions are never silently truncated.
payload_type = c_ubyte * native.entryLength
payload = adsSyncReadWriteReqEx2(
port,
self._adr,
ADSIGRP_SYM_INFOBYNAMEEX,
0,
payload_type,
name,
PLCTYPE_STRING,
return_ctypes=True,
check_length=True,
)
return parse_symbol_entry(payload, encoding=upload.symbol_encoding)
[docs]
def get_all_symbol_info(self) -> Tuple[AdsSymbolInfo, ...]:
"""Upload and parse every exported symbol as immutable metadata."""
self._require_open_port()
# A caller performing an explicit discovery refresh must never retain
# handle/layout information from the preceding runtime generation.
self.clear_symbol_cache()
upload = self.get_upload_info()
if upload.symbol_bytes == 0:
if upload.symbol_count:
raise RuntimeError(
"Runtime reported symbols with an empty upload blob."
)
return ()
payload_type = c_ubyte * upload.symbol_bytes
payload = self.read(
ADSIGRP_SYM_UPLOAD,
ADSIOFFS_DEVDATA_ADSSTATE,
payload_type,
return_ctypes=True,
)
return parse_symbol_entries(
payload,
count=upload.symbol_count,
encoding=upload.symbol_encoding,
)
[docs]
def get_all_data_types(self) -> Tuple[AdsDataType, ...]:
"""Upload and parse the complete PLC datatype table."""
self._require_open_port()
upload = self.get_upload_info()
if upload.data_type_bytes == 0:
if upload.data_type_count:
raise RuntimeError(
"Runtime reported datatypes with an empty upload blob."
)
return ()
payload_type = c_ubyte * upload.data_type_bytes
payload = self.read(
ADSIGRP_SYM_DT_UPLOAD,
ADSIOFFS_DEVDATA_ADSSTATE,
payload_type,
return_ctypes=True,
)
return parse_data_type_entries(
payload,
count=upload.data_type_count,
encoding=upload.symbol_encoding,
)
[docs]
def get_all_symbols(self) -> List[AdsSymbol]:
"""Read all symbols from an ADS-device.
The legacy active-symbol API now delegates parsing to
:meth:`get_all_symbol_info`. New discovery code should prefer the
immutable metadata API so symbol methods cannot bypass async
serialization.
"""
if self._port is None:
return []
symbols = []
for info in self.get_all_symbol_info():
symbol = AdsSymbol(
plc=self,
name=info.name,
index_group=info.index_group,
index_offset=info.index_offset,
symbol_type=info.type_name,
comment=info.comment,
metadata=info,
)
symbols.append(symbol)
return symbols
@overload
def get_object(
self,
object_name: str,
method_separator: str = "#",
method_prefixes: Tuple[str, ...] = ("", "m_"),
method_return_types: Optional[Dict[str, Type["PLCDataType"]]] = None,
method_parameters: Optional[
Dict[str, Sequence[Type["PLCDataType"]]]
] = None,
) -> RpcObject: ...
@overload
def get_object(
self,
object_name: Type[RpcInterfaceT],
method_separator: str = "#",
method_prefixes: Tuple[str, ...] = ("", "m_"),
method_return_types: Optional[Dict[str, Type["PLCDataType"]]] = None,
method_parameters: Optional[
Dict[str, Sequence[Type["PLCDataType"]]]
] = None,
) -> RpcInterfaceT: ...
[docs]
def get_object(
self,
object_name: Union[str, Type[RpcInterfaceT]],
method_separator: str = "#",
method_prefixes: Tuple[str, ...] = ("", "m_"),
method_return_types: Optional[Dict[str, Type["PLCDataType"]]] = None,
method_parameters: Optional[
Dict[str, Sequence[Type["PLCDataType"]]]
] = None,
) -> Union[RpcObject, RpcInterfaceT]:
"""Create a PLC object proxy for native RPC method calls or typed interfaces.
Example:
rpc = plc.get_object(
"GVL.fbTestRemoteMethodCall",
method_return_types={"m_iSum": pyads.PLCTYPE_INT},
method_parameters={"m_iSum": [pyads.PLCTYPE_INT, pyads.PLCTYPE_INT]},
)
result = rpc.m_iSum(5, 5)
Method names are resolved in order using ``method_prefixes``.
The default supports both ``doCalc`` and ``m_doCalc`` style RPC names.
"""
interface_class: Optional[Type[RpcInterfaceT]] = (
object_name if inspect.isclass(object_name) else None
)
inferred_returns: Optional[Dict[str, Type["PLCDataType"]]] = None
inferred_parameters: Optional[Dict[str, Tuple[Type["PLCDataType"], ...]]] = None
if isinstance(object_name, str):
resolved_object_name = object_name
elif interface_class is not None:
interface_definition = resolve_rpc_interface_definition(interface_class)
if interface_definition.async_interface:
raise TypeError(
f"{interface_class.__name__} is decorated with @ads_async_path('...') "
"and must be used with AsyncConnection.get_async_object()."
)
if interface_definition.stepchain_methods:
raise TypeError(
f"{interface_class.__name__} declares @stepchain_start methods "
"and must be used with AsyncConnection.get_async_object()."
)
resolved_object_name = interface_definition.object_name
if interface_definition.method_return_types:
inferred_returns = dict(interface_definition.method_return_types)
if interface_definition.method_parameters:
inferred_parameters = dict(interface_definition.method_parameters)
else:
raise TypeError("object_name must be a string or an ads_path-decorated class.")
final_return_types: Optional[Dict[str, Type["PLCDataType"]]] = None
if inferred_returns:
final_return_types = inferred_returns
if method_return_types:
final_return_types = {**(final_return_types or {}), **method_return_types}
normalized_parameters: Optional[Dict[str, Tuple[Type["PLCDataType"], ...]]] = None
if method_parameters:
normalized_parameters = {
name: tuple(values)
for name, values in method_parameters.items()
}
if inferred_parameters:
normalized_parameters = {
**(inferred_parameters or {}),
**(normalized_parameters or {}),
}
rpc_object = RpcObject(
connection=self,
object_name=resolved_object_name,
method_separator=method_separator,
method_prefixes=method_prefixes,
method_return_types=final_return_types,
method_parameters=normalized_parameters,
)
if interface_class is not None:
return cast(RpcInterfaceT, rpc_object)
return rpc_object
[docs]
def call_rpc_method(
self,
method_name: str,
return_type: Optional[Type["PLCDataType"]] = None,
write_value: Any = None,
write_type: Optional[Type["PLCDataType"]] = None,
) -> Any:
"""Call a PLC RPC method by ADS handle and release the handle afterwards."""
handle = self.get_handle(method_name)
if handle is None:
return None
try:
return self.read_write(
ADSIGRP_SYM_VALBYHND,
handle,
return_type,
write_value,
write_type,
)
finally:
self.release_handle(handle)
[docs]
def get_handle(self, data_name: str) -> Optional[int]:
"""Get the handle of the PLC-variable, handles obtained using this
method should be released using method 'release_handle'.
:param string data_name: data name
:rtype: int
:return: int: PLC-variable handle
"""
if self._port is not None:
return adsGetHandle(self._port, self._adr, data_name)
return None
[docs]
def release_handle(self, handle: int) -> None:
""" Release handle of a PLC-variable.
:param int handle: handle of PLC-variable to be released
"""
if self._port is not None:
adsReleaseHandle(self._port, self._adr, handle)
[docs]
def read_by_name(
self,
data_name: str,
plc_datatype: Optional[Type["PLCDataType"]] = None,
return_ctypes: bool = False,
handle: Optional[int] = None,
check_length: bool = True,
cache_symbol_info: bool = True,
) -> Any:
"""Read data synchronous from an ADS-device from data name.
:param string data_name: data name, can be empty string if handle is used
:param Optional[Type["PLCDataType"]] plc_datatype: type of the data given to the PLC, according
to PLCTYPE constants, if None the datatype will be read from the target
with adsGetSymbolInfo (default: None)
:param bool return_ctypes: return ctypes instead of python types if True
(default: False)
:param int handle: PLC-variable handle, pass in handle if previously
obtained to speed up reading (default: None)
:param bool check_length: check whether the amount of bytes read matches the size
of the read data type (default: True)
:param bool cache_symbol_info: when True, symbol info will be cached for
future reading, only relevant if plc_datatype is None (default: True)
:return: value: **value**
"""
if not self._port:
return
if plc_datatype is None:
plc_datatype = self._query_plc_datatype_from_name(data_name,
cache_symbol_info)
return adsSyncReadByNameEx(
self._port,
self._adr,
data_name,
plc_datatype,
return_ctypes=return_ctypes,
handle=handle,
check_length=check_length,
)
[docs]
def read_list_by_name(
self,
data_names: List[str],
cache_symbol_info: bool = True,
ads_sub_commands: int = MAX_ADS_SUB_COMMANDS,
structure_defs: Optional[Dict[str, StructureDef]] = None,
) -> Dict[str, Any]:
"""Read a list of variables.
Will split the read into multiple ADS calls in chunks of ads_sub_commands by default.
MAX_ADS_SUB_COMMANDS comes from Beckhoff recommendation:
https://infosys.beckhoff.com/english.php?content=../content/1033/tc3_adsdll2/9007199379576075.html&id=9180083787138954512
:param List[str] data_names: list of variable names to be read
:param bool cache_symbol_info: when True, symbol info will be cached for future reading
:param int ads_sub_commands: Max number of ADS-Sub commands used to read the variables in a single ADS call.
A larger number can be used but may jitter the PLC execution!
:param Optional[Dict[str, StructureDef]] structure_defs: for structured variables, optional mapping of
data name to special tuple defining the structure and types contained within it according to PLCTYPE constants
:return adsSumRead: A dictionary containing variable names from data_names as keys and values read from PLC for each variable
:rtype: Dict[str, Any]
"""
if structure_defs is None:
structure_defs = {}
if cache_symbol_info:
new_items = [i for i in data_names if i not in self._symbol_info_cache]
new_cache = {
i: adsGetSymbolInfo(self._port, self._adr, i) for i in new_items
}
self._symbol_info_cache.update(new_cache)
data_symbols = {i: self._symbol_info_cache[i] for i in data_names}
else:
data_symbols = {
i: adsGetSymbolInfo(self._port, self._adr, i) for i in data_names
}
def sum_read(port: int, adr: AmsAddr, data_names: List[str],
data_symbols: Dict) -> Dict[str, str]:
result = adsSumRead(port, adr, data_names, data_symbols,
list(structure_defs.keys()),
self._runtime_string_encoding()) # type: ignore
for data_name, structure_def in structure_defs.items(): # type: ignore
if data_name in result:
result[data_name] = dict_from_bytes(result[data_name],
structure_def)
return result
if len(data_names) <= ads_sub_commands:
return sum_read(self._port, self._adr, data_names, data_symbols)
return_data: Dict[str, Any] = {}
for data_names_slice in _list_slice_generator(data_names, ads_sub_commands):
return_data.update(
sum_read(self._port, self._adr, data_names_slice, data_symbols)
)
return return_data
[docs]
def read_list_by_name_detailed(
self,
data_names: List[str],
cache_symbol_info: bool = True,
ads_sub_commands: int = MAX_ADS_SUB_COMMANDS,
structure_defs: Optional[Dict[str, StructureDef]] = None,
*,
symbol_metadata: Optional[Mapping[str, AdsSymbolInfo]] = None,
raw_data_names: Optional[Sequence[str]] = None,
) -> Dict[str, SumReadResult]:
"""Read a list while preserving native ADS codes per requested item.
When ``symbol_metadata`` is supplied, its keys must exactly match the
requested reads. The immutable uploaded addresses, sizes, types, and
flags are used directly without consulting or updating the symbol
cache and without resolving any name again. This lets callers batch
arbitrary padded structure roots as raw values while retaining the
exact metadata generation they already verified. Names listed in
``raw_data_names`` are returned as copied bytes even when their ADS
datatype has a built-in scalar decoder.
"""
port = self._require_open_port()
if ads_sub_commands <= 0:
raise ValueError("ads_sub_commands must be greater than zero")
structure_defs = structure_defs or {}
if isinstance(raw_data_names, (str, bytes, bytearray)):
raise TypeError("raw_data_names must be a sequence of symbol names")
try:
raw_names = tuple(raw_data_names or ())
except TypeError as exc:
raise TypeError(
"raw_data_names must be a sequence of symbol names"
) from exc
if any(not isinstance(name, str) or not name for name in raw_names):
raise TypeError("raw_data_names must contain non-empty strings")
if any("\x00" in name for name in raw_names):
raise ValueError("raw_data_names must not contain NUL")
raw_name_set = frozenset(raw_names)
unknown_raw_names = raw_name_set.difference(data_names)
if unknown_raw_names:
raise ValueError(
"raw_data_names must be a subset of the requested names: "
+ ", ".join(sorted(unknown_raw_names))
)
results: Dict[str, SumReadResult] = {}
data_symbols: Dict[str, Union[SAdsSymbolEntry, AdsSymbolInfo]] = {}
if symbol_metadata is not None:
data_symbols.update(
_validated_exact_symbol_metadata(tuple(data_names), symbol_metadata)
)
else:
for name in data_names:
try:
info = self._symbol_info_cache.get(name) if cache_symbol_info else None
if info is None:
info = adsGetSymbolInfo(port, self._adr, name)
if cache_symbol_info:
self._symbol_info_cache[name] = info
data_symbols[name] = info
except ADSError as exc:
results[name] = SumReadResult(
name=name,
success=False,
error_code=exc.err_code,
error_message=ERROR_CODES.get(exc.err_code, str(exc)),
)
supported_symbols: Dict[str, Union[SAdsSymbolEntry, AdsSymbolInfo]] = {}
for name, info in data_symbols.items():
unsupported = _sum_symbol_unsupported_reason(name, info)
if unsupported is None:
supported_symbols[name] = info
else:
results[name] = SumReadResult(
name=name,
success=False,
error_message=unsupported,
)
data_symbols = supported_symbols
readable = [name for name in data_names if name in data_symbols]
for names_slice in _list_slice_generator(readable, ads_sub_commands):
native = adsSumReadDetailed(
port,
self._adr,
names_slice,
data_symbols,
list(dict.fromkeys((*structure_defs.keys(), *raw_names))),
self._runtime_string_encoding(),
)
for name, (error_code, value) in native.items():
if error_code:
results[name] = SumReadResult(
name=name,
success=False,
error_code=error_code,
error_message=ERROR_CODES.get(
error_code, f"Unknown ADS error {error_code}"
),
)
continue
try:
if name in structure_defs and name not in raw_name_set:
value = dict_from_bytes(value, structure_defs[name])
except Exception as exc:
results[name] = SumReadResult(
name=name,
success=False,
error_message=f"Failed to decode ADS value: {exc}",
)
else:
results[name] = SumReadResult(
name=name, success=True, value=value
)
return results
[docs]
def read_structure_by_name(
self,
data_name: str,
structure_def: StructureDef,
array_size: Optional[int] = 1,
structure_size: Optional[int] = None,
handle: Optional[int] = None,
) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]:
"""Read a structure of multiple types.
:param string data_name: data name
:param tuple structure_def: special tuple defining the structure and
types contained within it according to PLCTYPE constants, must match
the structure defined in the PLC, PLC structure must be defined with
{attribute 'pack_mode' := '1'}
:param Optional[int] array_size: size of array if reading array of structure, defaults to 1
:param Optional[int] structure_size: size of structure if known by previous use of
size_of_structure, defaults to None
:param Optional[int] handle: PLC-variable handle, pass in handle if previously
obtained to speed up reading, defaults to None
:return: values_dict: ordered dictionary of all values corresponding to the structure
definition
Expected input example for structure_def:
.. code:: python
structure_def = (
('rVar', pyads.PLCTYPE_LREAL, 1),
('sVar', pyads.PLCTYPE_STRING, 2, 35),
('SVar1', pyads.PLCTYPE_STRING, 1),
('rVar1', pyads.PLCTYPE_REAL, 1),
('iVar', pyads.PLCTYPE_DINT, 1),
('iVar1', pyads.PLCTYPE_INT, 3),
)
# i.e ('Variable Name', variable type, arr size (1 if not array),
# length of string (if defined in PLC))
"""
if structure_size is None:
structure_size = size_of_structure(structure_def * array_size)
values = self.read_by_name(data_name, c_ubyte * structure_size, handle=handle)
if values is not None:
return dict_from_bytes(values, structure_def, array_size=array_size)
return None
[docs]
def write_by_name(
self,
data_name: str,
value: Any,
plc_datatype: Optional[Type["PLCDataType"]] = None,
handle: Optional[int] = None,
cache_symbol_info: bool = True,
) -> None:
"""Send data synchronous to an ADS-device from data name.
:param string data_name: data name, can be empty string if handle is used
:param value: value to write to the storage address of the PLC
:param int plc_datatype: type of the data given to the PLC, according
to PLCTYPE constants, if None the datatype will be read from the target
with adsGetSymbolInfo (default: None)
:param int handle: PLC-variable handle, pass in handle if previously
obtained to speed up writing (default: None)
:param bool cache_symbol_info: when True, symbol info will be cached for
future reading, only relevant if plc_datatype is None (default: True)
"""
if not self._port:
return
if plc_datatype is None:
plc_datatype = self._query_plc_datatype_from_name(data_name,
cache_symbol_info)
return adsSyncWriteByNameEx(
self._port, self._adr, data_name, value, plc_datatype, handle=handle
)
[docs]
def write_list_by_name(
self,
data_names_and_values: Dict[str, Any],
cache_symbol_info: bool = True,
ads_sub_commands: int = MAX_ADS_SUB_COMMANDS,
structure_defs: Optional[Dict[str, StructureDef]] = None,
) -> Dict[str, str]:
"""Write a list of variables.
Will split the write into multiple ADS calls in chunks of ads_sub_commands by default.
MAX_ADS_SUB_COMMANDS comes from Beckhoff recommendation:
https://infosys.beckhoff.com/english.php?content=../content/1033/tc3_adsdll2/9007199379576075.html&id=9180083787138954512
:param data_names_and_values: dictionary of variable names and their values to be written
:type data_names_and_values: dict[str, Any]
:param bool cache_symbol_info: when True, symbol info will be cached for future reading
:param int ads_sub_commands: Max number of ADS-Sub commands used to write the variables in a single ADS call.
A larger number can be used but may jitter the PLC execution!
:param dict structure_defs: for structured variables, optional mapping of
data name to special tuple defining the structure and
types contained within it according to PLCTYPE constants
:return adsSumWrite: A dictionary containing variable names from data_names as keys and values return codes for
each write operation from the PLC
:rtype: dict(str, str)
"""
if cache_symbol_info:
new_items = [
i
for i in data_names_and_values.keys()
if i not in self._symbol_info_cache
]
new_cache = {
i: adsGetSymbolInfo(self._port, self._adr, i) for i in new_items
}
self._symbol_info_cache.update(new_cache)
data_symbols = {
i: self._symbol_info_cache[i] for i in data_names_and_values
}
else:
data_symbols = {
i: adsGetSymbolInfo(self._port, self._adr, i)
for i in data_names_and_values.keys()
}
if structure_defs is None:
structure_defs = {}
else:
data_names_and_values = data_names_and_values.copy() # copy so the original does not get modified
for name, structure_def in structure_defs.items():
data_names_and_values[name] = bytes(
bytes_from_dict(data_names_and_values[name], structure_def)
)
structured_data_names = list(structure_defs.keys())
if len(data_names_and_values) <= ads_sub_commands:
return adsSumWrite(
self._port, self._adr, data_names_and_values, data_symbols,
structured_data_names, self._runtime_string_encoding()
)
return_data: Dict[str, str] = {}
for data_names_slice in _dict_slice_generator(data_names_and_values,
ads_sub_commands):
return_data.update(
adsSumWrite(self._port, self._adr, data_names_slice, data_symbols,
structured_data_names,
self._runtime_string_encoding())
)
return return_data
[docs]
def write_list_by_name_detailed(
self,
data_names_and_values: Dict[str, Any],
cache_symbol_info: bool = True,
ads_sub_commands: int = MAX_ADS_SUB_COMMANDS,
structure_defs: Optional[Dict[str, StructureDef]] = None,
*,
symbol_metadata: Optional[Mapping[str, AdsSymbolInfo]] = None,
) -> Dict[str, SumWriteResult]:
"""Write a list while preserving native ADS codes per requested item.
When ``symbol_metadata`` is supplied, its keys must exactly match the
requested writes and every value must be an immutable
:class:`AdsSymbolInfo`. Those approved addresses, sizes, types, and
flags are used directly without consulting or updating the symbol
cache and without resolving any name again.
"""
port = self._require_open_port()
if ads_sub_commands <= 0:
raise ValueError("ads_sub_commands must be greater than zero")
structure_defs = structure_defs or {}
results: Dict[str, SumWriteResult] = {}
data_symbols: Dict[str, Union[SAdsSymbolEntry, AdsSymbolInfo]] = {}
if symbol_metadata is not None:
data_symbols.update(
_validated_exact_symbol_metadata(
tuple(data_names_and_values),
symbol_metadata,
)
)
else:
for name in data_names_and_values:
try:
info = self._symbol_info_cache.get(name) if cache_symbol_info else None
if info is None:
info = adsGetSymbolInfo(port, self._adr, name)
if cache_symbol_info:
self._symbol_info_cache[name] = info
data_symbols[name] = info
except ADSError as exc:
results[name] = SumWriteResult(
name=name,
success=False,
error_code=exc.err_code,
error_message=ERROR_CODES.get(exc.err_code, str(exc)),
)
supported_symbols: Dict[str, Union[SAdsSymbolEntry, AdsSymbolInfo]] = {}
for name, info in data_symbols.items():
unsupported = _sum_symbol_unsupported_reason(name, info)
if unsupported is None:
supported_symbols[name] = info
else:
results[name] = SumWriteResult(
name=name,
success=False,
error_message=unsupported,
)
data_symbols = supported_symbols
writable = {
name: value
for name, value in data_names_and_values.items()
if name in data_symbols
}
for name, structure_def in structure_defs.items():
if name in writable:
writable[name] = bytes(
bytes_from_dict(writable[name], structure_def)
)
structured_names = list(structure_defs.keys())
for values_slice in _dict_slice_generator(writable, ads_sub_commands):
native = adsSumWriteDetailed(
port,
self._adr,
values_slice,
data_symbols,
structured_names,
self._runtime_string_encoding(),
)
for name, error_code in native.items():
results[name] = SumWriteResult(
name=name,
success=error_code == 0,
error_code=error_code,
error_message=(
None
if error_code == 0
else ERROR_CODES.get(
error_code, f"Unknown ADS error {error_code}"
)
),
)
return results
[docs]
def write_structure_by_name(
self,
data_name: str,
value: Union[Dict[str, Any], List[Dict[str, Any]]],
structure_def: StructureDef,
array_size: Optional[int] = 1,
structure_size: Optional[int] = None,
handle: Optional[int] = None,
) -> None:
"""Write a structure of multiple types.
:param str data_name: data name
:param Union[Dict[str, Any], List[Dict[str, Any]]] value: value to write to the storage address of the PLC
:param StructureDef structure_def: special tuple defining the structure and
types contained within it according to PLCTYPE constants, must match
the structure defined in the PLC, PLC structure must be defined with
{attribute 'pack_mode' := '1'}
:param Optional[int] array_size: size of array if writing array of structure, defaults to 1
:param Optional[int] structure_size: size of structure if known by previous use of
size_of_structure, defaults to None
:param Optional[int] handle: PLC-variable handle, pass in handle if previously
obtained to speed up reading, defaults to None
Expected input example for structure_def:
.. code:: python
structure_def = (
('rVar', pyads.PLCTYPE_LREAL, 1),
('sVar', pyads.PLCTYPE_STRING, 2, 35),
('sVar', pyads.PLCTYPE_STRING, 1),
('rVar1', pyads.PLCTYPE_REAL, 1),
('iVar', pyads.PLCTYPE_DINT, 1),
)
# i.e ('Variable Name', variable type, arr size (1 if not array),
# length of string (if defined in PLC))
"""
byte_values = bytes_from_dict(value, structure_def)
if structure_size is None:
structure_size = size_of_structure(structure_def * array_size)
return self.write_by_name(
data_name, byte_values, c_ubyte * structure_size, handle=handle
)
[docs]
def add_device_notification(
self,
data: Union[str, Tuple[int, int]],
attr: NotificationAttrib,
callback: Callable,
user_handle: Optional[int] = None,
) -> Optional[Tuple[int, int]]:
"""Add a device notification.
:param Union[str, Tuple[int, int] data: PLC storage address as string or Tuple with index group and offset
:param pyads.structs.NotificationAttrib attr: object that contains
all the attributes for the definition of a notification
:param callback: callback function that gets executed in the event of a notification
:param user_handle: optional user handle
:rtype: (int, int)
:returns: notification handle, user handle
Save the notification handle and the user handle on creating a
notification if you want to be able to remove the notification
later in your code.
**Usage**:
>>> import pyads
>>> from ctypes import sizeof
>>>
>>> # Connect to the local TwinCAT PLC
>>> plc = pyads.Connection('127.0.0.1.1.1', 851)
>>>
>>> # Create callback function that prints the value
>>> def mycallback(notification, data):
>>> contents = notification.contents
>>> value = next(
>>> map(int,
>>> bytearray(contents.data)[0:contents.cbSampleSize])
>>> )
>>> print(value)
>>>
>>> with plc:
>>> # Add notification with default settings
>>> atr = pyads.NotificationAttrib(sizeof(pyads.PLCTYPE_INT))
>>> handles = plc.add_device_notification("GVL.myvalue", atr, mycallback)
>>>
>>> # Remove notification
>>> plc.del_device_notification(handles)
Note: the `user_handle` (passed or returned) is the same as the handle returned from
:meth:`Connection.get_handle()`.
"""
if self._port is not None:
notification_handle, user_handle = adsSyncAddDeviceNotificationReqEx(
self._port, self._adr, data, attr, callback, user_handle
)
return notification_handle, user_handle
return None
[docs]
def del_device_notification(
self, notification_handle: int, user_handle: Optional[int]
) -> None:
"""Remove a device notification.
:param notification_handle: address of the variable that contains
the handle of the notification
:param user_handle: user handle
"""
if self._port is not None:
adsSyncDelDeviceNotificationReqEx(
self._port, self._adr, notification_handle, user_handle
)
@property
def is_open(self) -> bool:
"""Show the current connection state.
:return: True if connection is open
"""
return self._open
[docs]
def set_timeout(self, ms: int) -> None:
"""Set Timeout."""
if self._port is not None:
adsSyncSetTimeoutEx(self._port, ms)
[docs]
def notification(
self, plc_datatype: Optional[Type] = None,
timestamp_as_filetime: bool = False
) -> Callable:
"""Decorate a callback function.
**Decorator**.
A decorator that can be used for callback functions in order to
convert the data of the NotificationHeader into the fitting
Python type.
:param plc_datatype: The PLC datatype that needs to be converted. This can
be any basic PLC datatype or a `ctypes.Structure`.
:param timestamp_as_filetime: Whether the notification timestamp should be returned
as `datetime.datetime` (False) or Windows `FILETIME` as originally transmitted
via ADS (True). Be aware that the precision of `datetime.datetime` is limited to
microseconds, while FILETIME allows for 100 ns. This may be relevant when using
task cycle times such as 62.5 µs. Default: False.
The callback functions need to be of the following type:
>>> def callback(handle, name, timestamp, value)
* `handle`: the notification handle
* `name`: the variable name
* `timestamp`: the timestamp as datetime value
* `value`: the converted value of the variable
**Usage**:
>>> import pyads
>>>
>>> plc = pyads.Connection('172.18.3.25.1.1', 851)
>>>
>>>
>>> @plc.notification(pyads.PLCTYPE_STRING)
>>> def callback(handle, name, timestamp, value):
>>> print(handle, name, timestamp, value)
>>>
>>>
>>> with plc:
>>> attr = pyads.NotificationAttrib(20,
>>> pyads.ADSTRANS_SERVERCYCLE)
>>> handles = plc.add_device_notification('GVL.test', attr,
>>> callback)
>>> while True:
>>> pass
"""
def notification_decorator(
func: Callable[[int, str, Union[datetime, int], Any], None]
) -> Callable[[Any, str], None]:
def func_wrapper(notification: Any, data_name: str) -> None:
h_notification, timestamp, value = self.parse_notification(
notification, plc_datatype, timestamp_as_filetime
)
return func(h_notification, data_name, timestamp, value)
return func_wrapper
return notification_decorator
# noinspection PyMethodMayBeStatic
[docs]
def parse_notification(
self,
notification: Any,
plc_datatype: Optional[Type],
timestamp_as_filetime: bool = False,
) -> Tuple[int, Union[datetime, int], Any]:
# noinspection PyTypeChecker
"""Parse a notification.
Convert the data of the NotificationHeader into the fitting Python type.
:param notification: The notification we recieve from PLC datatype to be
converted. This can be any basic PLC datatype or a `ctypes.Structure`.
:param plc_datatype: The PLC datatype that needs to be converted. This can
be any basic PLC datatype or a `ctypes.Structure`.
:param timestamp_as_filetime: Whether the notification timestamp should be returned
as `datetime.datetime` (False) or Windows `FILETIME` as originally transmitted
via ADS (True). Be aware that the precision of `datetime.datetime` is limited to
microseconds, while FILETIME allows for 100 ns. This may be relevant when using
task cycle times such as 62.5 µs. Default: False.
:rtype: (int, int, Any)
:returns: notification handle, timestamp, value
**Usage**:
>>> import pyads
>>> from ctypes import sizeof
>>>
>>> # Connect to the local TwinCAT PLC
>>> plc = pyads.Connection('127.0.0.1.1.1', 851)
>>> tag = {"GVL.myvalue": pyads.PLCTYPE_INT}
>>>
>>> # Create callback function that prints the value
>>> def mycallback(notification: SAdsNotificationHeader, data: str) -> None:
>>> data_type = tag[data]
>>> handle, timestamp, value = plc.parse_notification(notification, data_type)
>>> print(value)
>>>
>>> with plc:
>>> # Add notification with default settings
>>> attr = pyads.NotificationAttrib(sizeof(pyads.PLCTYPE_INT))
>>>
>>> handles = plc.add_device_notification("GVL.myvalue", attr, mycallback)
>>>
>>> # Remove notification
>>> plc.del_device_notification(handles)
"""
contents = notification.contents
data_size = contents.cbSampleSize
# Get dynamically sized data array
data = (c_ubyte * data_size).from_address(
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")
elif plc_datatype is not None and issubclass(plc_datatype, Structure):
value = plc_datatype()
fit_size = min(data_size, sizeof(value))
memmove(addressof(value), addressof(data), fit_size)
elif plc_datatype is not None and issubclass(plc_datatype, Array):
if data_size == sizeof(plc_datatype):
value = list(plc_datatype.from_buffer_copy(bytes(data)))
else:
# invalid size
value = None
elif plc_datatype not in DATATYPE_MAP:
value = bytearray(data)
else:
value = struct.unpack(DATATYPE_MAP[plc_datatype], bytearray(data))[0]
if timestamp_as_filetime:
timestamp = contents.nTimeStamp
else:
timestamp = filetime_to_dt(contents.nTimeStamp)
return contents.hNotification, timestamp, value