pyads-agile package (import name: pyads)

The submodules of the pyads-agile package are listed below.

pyads.ads module

Pythonic ADS functions.

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

pyads.ads.add_route(adr: str | AmsAddr | None, ip_address: str) → None[source]

Establish a new route in the AMS Router (linux Only).

Parameters:

adr – AMS Address of routing endpoint as str or AmsAddr object. If None is provided, the net id of the PLC will be discovered.
ip_address (str) – ip address of the routing endpoint

pyads.ads.add_route_to_plc(sending_net_id: str, adding_host_name: str, ip_address: str, username: str, password: str, route_name: str = None, added_net_id: str = None) → bool[source]

Embed a new route in the PLC.

Parameters:

sending_net_id (str) – sending net id
adding_host_name (str) – host name (or IP) of the PC being added
ip_address (str) – ip address of the PLC
username (str) – username for PLC
password (str) – password for PLC
route_name (str) – PLC side name for route, defaults to adding_host_name or the current hostname of this PC
added_net_id (pyads.structs.SAmsNetId) – net id that is being added to the PLC, defaults to sending_net_id

Return type:

bool

Returns:

True if route was added

pyads.ads.bytes_from_dict(values: Dict[str, Any] | List[Dict[str, Any]], structure_def: Tuple[Tuple[str, Type, int] | Tuple[str, Type, int, int | None], ...]) → List[int][source]

Returns a byte array of values which can be written to the PLC from an ordered dict.

Parameters:

values – ordered dictionary of values for each variable type in order of structure_def
structure_def (tuple) – special tuple defining the structure and types contained within it according o PLCTYPE constants
array_size (Optional[int]) – size of array if writing array of structure, defaults to 1

Returns:

list of byte values for an entire structure

Return type:

List[int]

Expected input example for structure_def:

structure_def = (
    ('rVar', pyads.PLCTYPE_LREAL, 1),
    ('sVar', pyads.PLCTYPE_STRING, 2, 35),
    ('sVar2', 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))

pyads.ads.close_port() → None[source]: Close the connection to the TwinCAT message router.

pyads.ads.delete_route(adr: AmsAddr) → None[source]

Remove existing route from the AMS Router (Linux Only).

Parameters:: adr (pyads.structs.AmsAddr) – AMS Address associated with the routing entry which is to be removed from the router.

pyads.ads.dict_from_bytes(byte_list: bytearray, structure_def: Tuple[Tuple[str, Type, int] | Tuple[str, Type, int, int | None], ...], array_size: int = 1) → Dict[str, Any] | List[Dict[str, Any]][source]

Return an ordered dict of PLC values from a list of BYTE values read from PLC.

Parameters:

byte_list (bytearray) – list of byte values for an entire structure
structure_def (tuple) – special tuple defining the structure and types contained within it according o PLCTYPE constants
array_size (Optional[int]) – size of array if reading array of structure, defaults to 1

Returns:

ordered dictionary of values for each variable type in order of structure

Expected input example for structure_def:

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))

pyads.ads.get_local_address() → AmsAddr | None[source]

Return the local AMS-address and the port number.

Return type:: AmsAddr

pyads.ads.open_port() → int[source]

Connect to the TwinCAT message router.

Return type:: int
Returns:: port number

pyads.ads.set_local_address(ams_netid: str | SAmsNetId) → None[source]

Set the local NetID (Linux only).

Parameters:: ams_netid (str) – new AmsNetID
Return type:: None

Usage:

>>> import pyads
>>> pyads.open_port()
>>> pyads.set_local_address('0.0.0.0.1.1')

pyads.ads.set_timeout(ms: int) → None[source]: Set timeout.

pyads.ads.size_of_structure(structure_def: Tuple[Tuple[str, Type, int] | Tuple[str, Type, int, int | None], ...]) → int[source]

Calculate the size of a structure in number of BYTEs.

Parameters:: structure_def (tuple) – special tuple defining the structure and types contained within it according o PLCTYPE constants
Returns:: data size required to read/write a structure of multiple types
Return type:: int

Expected input example for structure_def:

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 array of structure multiply structure_def input by array size.

pyads.constants module

Constants for the work with the ADS API.

author:: Stefan Lehmann <stlm@posteo.de>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2018-06-11 18:15:53

pyads.constants.ADSIGRP_DEVICE_DATA = 61696: state, name, etc…

pyads.constants.ADSIGRP_IOIMAGE_CLEARI = 61504: write inputs to null

pyads.constants.ADSIGRP_IOIMAGE_CLEARO = 61520: write outputs to null

pyads.constants.ADSIGRP_IOIMAGE_RWIB = 61472: read/write input byte(s)

pyads.constants.ADSIGRP_IOIMAGE_RWIX = 61473: read/write input bit

pyads.constants.ADSIGRP_IOIMAGE_RWOB = 61488: read/write output byte(s)

pyads.constants.ADSIGRP_IOIMAGE_RWOX = 61489: read/write output bit

pyads.constants.ADSIGRP_SUMUP_READ = 61568: ADS Sum Read Request

pyads.constants.ADSIGRP_SUMUP_WRITE = 61569: ADS Sum Write Request

pyads.constants.ADSIGRP_SYMNOTE = 61456: notification of named handle

pyads.constants.ADSIOFFS_DEVDATA_ADSSTATE = 0: ads state of device

pyads.constants.ADSIOFFS_DEVDATA_DEVSTATE = 2: device state

pyads.constants.ADSTRANS_NOTRANS: int = 0: no notifications

pyads.constants.ADSTRANS_SERVERCYCLE: int = 3: notify on a cyclic base

pyads.constants.ADSTRANS_SERVERONCHA: int = 4: notify everytime the value changes

pyads.constants.INDEXGROUP_DATA = 16448: data area, offset means byte-offset

pyads.constants.INDEXGROUP_DATASIZE = 16453: size of the data area in bytes

pyads.constants.INDEXGROUP_MEMORYBIT = 16417: plc memory area (%MX), offset means the bit address, calculatedb by bytenumber * 8 + bitnumber # noqa: E501

pyads.constants.INDEXGROUP_MEMORYBYTE = 16416: plc memory area (%M), offset means byte-offset

pyads.constants.INDEXGROUP_MEMORYSIZE = 16421: size of the memory area in bytes

pyads.constants.INDEXGROUP_RETAIN = 16432: plc retain memory area, offset means byte-offset

pyads.constants.INDEXGROUP_RETAINSIZE = 16437: size of the retain area in bytes

pyads.constants.PLCTYPE_ARR_BOOL(n: int) → Type[Array][source]: Return an array with n boolean values.

pyads.constants.PLCTYPE_ARR_DINT(n: int) → Type[Array][source]: Return an array with n int32 values.

pyads.constants.PLCTYPE_ARR_INT(n: int) → Type[Array][source]: Return an array with n int16 values.

pyads.constants.PLCTYPE_ARR_LREAL(n: int) → Type[Array][source]: Return an array with n double values.

pyads.constants.PLCTYPE_ARR_REAL(n: int) → Type[Array][source]: Return an array with n float values.

pyads.constants.PLCTYPE_ARR_SHORT(n: int) → Type[Array][source]: Return an array with n short values.

pyads.constants.PLCTYPE_ARR_SINT(n: int) → Type[Array][source]: Return an array with n int8 values.

pyads.constants.PLCTYPE_ARR_UDINT(n: int) → Type[Array][source]: Return an array with n uint32 values.

pyads.constants.PLCTYPE_ARR_UINT(n: int) → Type[Array][source]: Return an array with n uint16 values.

pyads.constants.PLCTYPE_ARR_USHORT(n: int) → Type[Array][source]: Return an array with n ushort values.

pyads.constants.PLCTYPE_ARR_USINT(n: int) → Type[Array][source]: Return an array with n uint8 values.

class pyads.constants.PLCTYPE_WSTRING[source]

Bases: object

Special dummy class for handling WSTRING.

pyads.connection module

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

class pyads.connection.Connection(ams_net_id: str = None, ams_net_port: int = None, ip_address: str = None)[source]

Bases: object

Class for managing the connection to an ADS device.

Variables:

ams_net_id (str) – AMS net id of the remote device
ams_net_port (int) – port of the remote device
ip_address (str) – 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.

add_device_notification(data: str | Tuple[int, int], attr: NotificationAttrib, callback: Callable, user_handle: int | None = None) → Tuple[int, int] | None[source]

Add a device notification.

Parameters:

data (Union[str, Tuple[int, int]) – PLC storage address as string or Tuple with index group and offset
attr (pyads.structs.NotificationAttrib) – object that contains all the attributes for the definition of a notification
callback – callback function that gets executed in the event of a notification
user_handle – optional user handle

Return type:

(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 Connection.get_handle().

property ams_netid: str

property ams_port: int

call_rpc_method(method_name: str, return_type: Type[Array | c_bool | c_ubyte | c_uint | c_int | c_short | c_double | c_float | c_byte | c_char | PLCTYPE_WSTRING | c_ushort | c_long | c_ulong] | None = None, write_value: Any = None, write_type: Type[Array | c_bool | c_ubyte | c_uint | c_int | c_short | c_double | c_float | c_byte | c_char | PLCTYPE_WSTRING | c_ushort | c_long | c_ulong] | None = None) → Any[source]: Call a PLC RPC method by ADS handle and release the handle afterwards.

close() → None[source]

Summary:: Close the connection to the TwinCAT message router.

del_device_notification(notification_handle: int, user_handle: int) → None[source]

Remove a device notification.

Parameters:

notification_handle – address of the variable that contains the handle of the notification
user_handle – user handle

get_all_symbols() → List[AdsSymbol][source]

Read all symbols from an ADS-device.

Returns:: List of AdsSymbols

get_handle(data_name: str) → int | None[source]

Get the handle of the PLC-variable, handles obtained using this: method should be released using method ‘release_handle’.

Parameters:: data_name (string) – data name
Return type:: int
Returns:: int: PLC-variable handle

get_local_address() → AmsAddr | None[source]

Return the local AMS-address and the port number.

Return type:: AmsAddr

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.

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’).

Parameters:

name (str)
index_group (Optional[int])
index_offset (Optional[int])
plc_datatype – type of the PLC variable, according to PLCTYPE constants
comment (str) – comment
auto_update (bool) – Create notification to update buffer (same as set_auto_update(True))
structure_def (Optional["StructureDef"]) – 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’}
array_size (Optional[int]) – size of array if reading array of structure, defaults to 1

Expected input example for structure_def:

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))

property is_open: bool

Show the current connection state.

Returns:: True if connection is open

notification(plc_datatype: Type | None = None, timestamp_as_filetime: bool = False) → Callable[source]

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.

Parameters:

plc_datatype – The PLC datatype that needs to be converted. This can be any basic PLC datatype or a ctypes.Structure.
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

open() → None[source]: Connect to the TwinCAT message router.

parse_notification(notification: Any, plc_datatype: Type | None, timestamp_as_filetime: bool = False) → Tuple[int, datetime | int, Any][source]

Parse a notification.

Convert the data of the NotificationHeader into the fitting Python type.

Parameters:

notification – The notification we recieve from PLC datatype to be converted. This can be any basic PLC datatype or a ctypes.Structure.
plc_datatype – The PLC datatype that needs to be converted. This can be any basic PLC datatype or a ctypes.Structure.
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.

Return type:

(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)

Read data synchronous from an ADS-device.

Parameters:

index_group (int) – PLC storage area, according to the INDEXGROUP constants
index_offset (int) – PLC storage address
plc_datatype (Type["PLCDataType"]) – type of the data given to the PLC, according to PLCTYPE constants
return_ctypes (bool) – return ctypes instead of python types if True (default: False)
check_length (bool) – check whether the amount of bytes read matches the size of the read data type (default: True)

Returns:

value

Read data synchronous from an ADS-device from data name.

Parameters:

data_name (string) – data name, can be empty string if handle is used
plc_datatype (Optional[Type["PLCDataType"]]) – 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)
return_ctypes (bool) – return ctypes instead of python types if True (default: False)
handle (int) – PLC-variable handle, pass in handle if previously obtained to speed up reading (default: None)
check_length (bool) – check whether the amount of bytes read matches the size of the read data type (default: True)
cache_symbol_info (bool) – when True, symbol info will be cached for future reading, only relevant if plc_datatype is None (default: True)

Returns:

value: value

read_device_info() → Tuple[str, AdsVersion] | None[source]

Read the name and the version number of the ADS-server.

Return type:: string, AdsVersion
Returns:: device name, version

read_list_by_name(data_names: List[str], cache_symbol_info: bool = True, ads_sub_commands: int = 500, structure_defs: Dict[str, Tuple[Tuple[str, Type, int] | Tuple[str, Type, int, int | None], ...]] | None = None) → Dict[str, Any][source]

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

Parameters:

data_names (List[str]) – list of variable names to be read
cache_symbol_info (bool) – when True, symbol info will be cached for future reading
ads_sub_commands (int) – 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!
structure_defs (Optional[Dict[str, StructureDef]]) – 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

Return type:

Dict[str, Any]

read_state() → Tuple[int, int] | None[source]

Read the current ADS-state and the machine-state.

Read the current ADS-state and the machine-state from the ADS-server.

Return type:: (int, int)
Returns:: adsState, deviceState

Read a structure of multiple types.

Parameters:

data_name (string) – data name
structure_def (tuple) – 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’}
array_size (Optional[int]) – size of array if reading array of structure, defaults to 1
structure_size (Optional[int]) – size of structure if known by previous use of size_of_structure, defaults to None
handle (Optional[int]) – PLC-variable handle, pass in handle if previously obtained to speed up reading, defaults to None

Returns:

values_dict: ordered dictionary of all values corresponding to the structure definition

Expected input example for structure_def:

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))

Read and write data synchronous from/to an ADS-device.

Parameters:

index_group (int) – PLC storage area, according to the INDEXGROUP constants
index_offset (int) – PLC storage address
plc_read_datatype (Type["PLCDataType"]) – type of the data given to the PLC to respond to, according to PLCTYPE constants, or None to not read anything
value – value to write to the storage address of the PLC
plc_write_datatype (Type["PLCDataType"]) – type of the data given to the PLC, according to PLCTYPE constants, or None to not write anything
return_ctypes (bool) – return ctypes instead of python types if True (default: False)
check_length (bool) – check whether the amount of bytes read matches the size of the read data type (default: True)

Returns:

value: value

release_handle(handle: int) → None[source]

Release handle of a PLC-variable.

Parameters:: handle (int) – handle of PLC-variable to be released

set_timeout(ms: int) → None[source]: Set Timeout.

Send data synchronous to an ADS-device.

Parameters:

index_group (int) – PLC storage area, according to the INDEXGROUP constants
index_offset (int) – PLC storage address
value (Any) – value to write to the storage address of the PLC
plc_datatype (Type["PLCDataType"]) – type of the data given to the PLC, according to PLCTYPE constants

Send data synchronous to an ADS-device from data name.

Parameters:

data_name (string) – data name, can be empty string if handle is used
value – value to write to the storage address of the PLC
plc_datatype (int) – 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)
handle (int) – PLC-variable handle, pass in handle if previously obtained to speed up writing (default: None)
cache_symbol_info (bool) – when True, symbol info will be cached for future reading, only relevant if plc_datatype is None (default: True)

write_control(ads_state: int, device_state: int, data: Any, plc_datatype: Type) → None[source]

Change the ADS state and the machine-state of the ADS-server.

Parameters:

ads_state (int) – new ADS-state, according to ADSTATE constants
device_state (int) – new machine-state
data – additional data
plc_datatype (int) – 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.

write_list_by_name(data_names_and_values: Dict[str, Any], cache_symbol_info: bool = True, ads_sub_commands: int = 500, structure_defs: Dict[str, Tuple[Tuple[str, Type, int] | Tuple[str, Type, int, int | None], ...]] | None = None) → Dict[str, str][source]

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

Parameters:

data_names_and_values (dict[str, Any]) – dictionary of variable names and their values to be written
cache_symbol_info (bool) – when True, symbol info will be cached for future reading
ads_sub_commands (int) – 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!
structure_defs (dict) – 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

Return type:

dict(str, str)

Write a structure of multiple types.

Parameters:

data_name (str) – data name
value (Union[Dict[str, Any], List[Dict[str, Any]]]) – value to write to the storage address of the PLC
structure_def (StructureDef) – 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’}
array_size (Optional[int]) – size of array if writing array of structure, defaults to 1
structure_size (Optional[int]) – size of structure if known by previous use of size_of_structure, defaults to None
handle (Optional[int]) – PLC-variable handle, pass in handle if previously obtained to speed up reading, defaults to None

Expected input example for structure_def:

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))

Bases: object

Proxy object for calling PLC RPC methods using native attribute syntax.

set_return_type(method_name: str, plc_type: Type[Array | c_bool | c_ubyte | c_uint | c_int | c_short | c_double | c_float | c_byte | c_char | PLCTYPE_WSTRING | c_ushort | c_long | c_ulong]) → None[source]: Register or override the expected return type for a method.

pyads.errorcodes module

Error codes.

author:: Stefan Lehmann <stlm@posteo.de>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2018-06-11 18:15:53

pyads.pyads_ex module

Contains cross platform ADS extension functions.

author:: David Browne <davidabrowne@gmail.com>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2018-06-11 18:15:53

exception pyads.pyads_ex.ADSError(err_code: int | None = None, text: str | None = None)[source]

Bases: Exception

Error class for errors related to ADS communication.

pyads.pyads_ex.adsAddRoute(net_id: SAmsNetId, ip_address: str) → None[source]

Establish a new route in the AMS Router.

Parameters:

net_id (pyads.structs.SAmsNetId) – net id of routing endpoint
ip_address (str) – ip address of the routing endpoint

pyads.pyads_ex.adsAddRouteToPLC(sending_net_id: str, adding_host_name: str, ip_address: str, username: str, password: str, route_name: str | None = None, added_net_id: str | None = None) → bool[source]

Embed a new route in the PLC.

Parameters:

sending_net_id (pyads.structs.SAmsNetId) – sending net id
adding_host_name (str) – host name (or IP) of the PC being added
ip_address (str) – ip address of the PLC
username (str) – username for PLC
password (str) – password for PLC
route_name (str) – PLC side name for route, defaults to adding_host_name or the current hostname of this PC
added_net_id (pyads.structs.SAmsNetId) – net id that is being added to the PLC, defaults to sending_net_id

Return type:

bool

Returns:

True if the provided credentials are correct, False otherwise

pyads.pyads_ex.adsDelRoute(net_id: SAmsNetId) → None[source]

Remove existing route from the AMS Router.

Parameters:: net_id (pyads.structs.SAmsNetId) – net id associated with the routing entry which is to be removed from the router.

pyads.pyads_ex.adsGetHandle(port: int, address: AmsAddr, data_name: str) → int[source]

Get the handle of the PLC-variable.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
data_name (string) – data name

Return type:

int

Returns:

handle: PLC-variable handle

pyads.pyads_ex.adsGetLocalAddressEx(port: int) → AmsAddr[source]

Return the local AMS-address and the port number.

Return type:: pyads.structs.AmsAddr
Returns:: AMS-address

pyads.pyads_ex.adsGetNetIdForPLC(ip_address: str) → str[source]

Get AMS Net ID from IP address.

Parameters:: ip_address (str) – ip address of the PLC
Return type:: str
Returns:: net id of the device at the provided ip address

pyads.pyads_ex.adsGetSymbolInfo(port: int, address: AmsAddr, data_name: str) → SAdsSymbolEntry[source]

Get the symbol information of the PLC-variable.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
data_name (string) – data name

Return type:

SAdsSymbolInfo

Returns:

symbol_info: PLC Symbol info

pyads.pyads_ex.adsPortCloseEx(port: int) → None[source]: Close the connection to the TwinCAT message router.

pyads.pyads_ex.adsPortOpenEx() → int[source]

Connect to the TwinCAT message router.

Return type:: int
Returns:: port number

pyads.pyads_ex.adsReleaseHandle(port: int, address: AmsAddr, handle: int) → None[source]

Release the handle of the PLC-variable.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
handle (int) – handle of PLC-variable to be released

pyads.pyads_ex.adsSetLocalAddress(ams_netid: SAmsNetId) → None[source]

Change the local NetId.

Parameters:: ams_netid (pyads.structs.SAmsNetId) – new AmsNetID
Return type:: None

pyads.pyads_ex.adsSumRead(port: int, address: AmsAddr, data_names: List[str], data_symbols: Dict[str, SAdsSymbolEntry], structured_data_names: List[str]) → Dict[str, Any][source]

Perform a sum read to get the value of multiple variables

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
data_names – list of variables names to read
data_symbols (Dict[str, SAdsSymbolEntry]) – dictionary of ADS Symbol Info
structured_data_names – list of structured variable names

Returns:

result: dict of variable names and values

Return type:

dict[str, Any]

pyads.pyads_ex.adsSumReadBytes(port: int, address: AmsAddr, data_symbols: List[Tuple[int, int, int]]) → Any[source]

Perform a sum read for multiple variables, returning the bytes

This version does not do any processing, and will simply return the concatenation of the bytes of the target symbols.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
data_symbols – list of tuples like: (index_group, index_offset, size)

pyads.pyads_ex.adsSumWrite(port: int, address: AmsAddr, data_names_and_values: Dict[str, Any], data_symbols: Dict[str, SAdsSymbolEntry], structured_data_names: List[str]) → Dict[str, str][source]

Perform a sum write to write the value of multiple ADS variables

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
data_names_and_values (dict[str, Any]) – dict of variable names and values to be written
data_symbols (dict[str, ADSSymbolInfo]) – list of dictionaries of ADS Symbol Info
structured_data_names – list of structured variable names

Returns:

result: dict of variable names and error codes

Return type:

dict[str, ADSError]

pyads.pyads_ex.adsSumWriteBytes(port: int, address: AmsAddr, num_requests: int, buffer: bytes) → List[str][source]

Perform a sum write of concatenated bytes to multiple symbols.

Returns:: List of errors

pyads.pyads_ex.adsSyncAddDeviceNotificationReqEx(port: int, adr: AmsAddr, data: str | Tuple[int, int], pNoteAttrib: NotificationAttrib, callback: Callable, user_handle: int | None = None) → Tuple[int, int][source]

Add a device notification.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
adr (pyads.structs.AmsAddr) – local or remote AmsAddr
data (Union[str, Tuple[int, int]]) – PLC storage address by name or index group and offset
pNoteAttrib (pyads.structs.NotificationAttrib) – notification attributes
callback – Callback function to handle notification
user_handle – User Handle

Return type:

(int, int)

Returns:

notification handle, user handle

pyads.pyads_ex.adsSyncDelDeviceNotificationReqEx(port: int, adr: AmsAddr, notification_handle: int, user_handle: int) → None[source]

Remove a device notification.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
adr (pyads.structs.AmsAddr) – local or remote AmsAddr
notification_handle (int) – Notification Handle
user_handle (int) – User Handle

pyads.pyads_ex.adsSyncReadByNameEx(port: int, address: AmsAddr, data_name: str, data_type: Type, return_ctypes: bool = False, handle: int = None, check_length: bool = True) → Any[source]

Read data synchronous from an ADS-device from data name.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
data_name (string) – data name
data_type (Type) – type of the data given to the PLC, according to PLCTYPE constants
return_ctypes (bool) – return ctypes instead of python types if True (default: False)
handle (int) – PLC-variable handle (default: None)
check_length (bool) – check whether the amount of bytes read matches the size of the read data type (default: True)

Return type:

data_type

Returns:

value: value

pyads.pyads_ex.adsSyncReadDeviceInfoReqEx(port: int, address: AmsAddr) → Tuple[str, AdsVersion][source]

Read the name and the version number of the ADS-server.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr

Return type:

string, AdsVersion

Returns:

device name, version

pyads.pyads_ex.adsSyncReadReqEx2(port: int, address: AmsAddr, index_group: int, index_offset: int, data_type: Type, return_ctypes: bool = False, check_length: bool = True) → Any[source]

Read data synchronous from an ADS-device.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
index_group (int) – PLC storage area, according to the INDEXGROUP constants
index_offset (int) – PLC storage address
data_type (Type) – type of the data given to the PLC, according to PLCTYPE constants
return_ctypes (bool) – return ctypes instead of python types if True (default: False)
check_length (bool) – check whether the amount of bytes read matches the size of the read data type (default: True)

Return type:

data_type

Returns:

value: value

pyads.pyads_ex.adsSyncReadStateReqEx(port: int, address: AmsAddr) → Tuple[int, int][source]

Read the current ADS-state and the machine-state.

Read the current ADS-state and the machine-state from the ADS-server.

Parameters:

port – AMS port
address (pyads.structs.AmsAddr) – local or remote AmsAddr

Return type:

(int, int)

Returns:

ads_state, device_state

pyads.pyads_ex.adsSyncReadWriteReqEx2(port: int, address: AmsAddr, index_group: int, index_offset: int, read_data_type: Type | None, value: Any, write_data_type: Type | None, return_ctypes: bool = False, check_length: bool = True) → Any[source]

Read and write data synchronous from/to an ADS-device.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
index_group (int) – PLC storage area, according to the INDEXGROUP constants
index_offset (int) – PLC storage address
read_data_type (Type) – type of the data given to the PLC to respond to, according to PLCTYPE constants, or None to not read anything
value – value to write to the storage address of the PLC
write_data_type (Type) – type of the data given to the PLC, according to PLCTYPE constants, or None to not write anything
return_ctypes (bool) – return ctypes instead of python types if True (default: False)
check_length (bool) – check whether the amount of bytes read matches the size of the read data type (default: True)

Return type:

read_data_type

Returns:

value: value read from PLC

pyads.pyads_ex.adsSyncSetTimeoutEx(port: int, n_ms: int) → None[source]

Set Timeout.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
n_ms (int) – timeout in ms

pyads.pyads_ex.adsSyncWriteByNameEx(port: int, address: AmsAddr, data_name: str, value: Any, data_type: Type, handle: int = None) → None[source]

Send data synchronous to an ADS-device from data name.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
data_name (string) – PLC storage name
value – value to write to the storage address of the PLC
data_type (Type) – type of the data given to the PLC, according to PLCTYPE constants
handle (int) – PLC-variable handle (default: None)

pyads.pyads_ex.adsSyncWriteControlReqEx(port: int, address: AmsAddr, ads_state: int, device_state: int, data: Any, plc_data_type: Type) → None[source]

Change the ADS state and the machine-state of the ADS-server.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
ads_state (int) – new ADS-state, according to ADSTATE constants
device_state (int) – new machine-state
data – additional data
plc_data_type (int) – plc datatype, according to PLCTYPE constants

pyads.pyads_ex.adsSyncWriteReqEx(port: int, address: AmsAddr, index_group: int, index_offset: int, value: Any, plc_data_type: Type) → None[source]

Send data synchronous to an ADS-device.

Parameters:

port (int) – local AMS port as returned by adsPortOpenEx()
address (pyads.structs.AmsAddr) – local or remote AmsAddr
index_group (int) – PLC storage area, according to the INDEXGROUP constants
index_offset (int) – PLC storage address
value – value to write to the storage address of the PLC
plc_data_type (int) – type of the data given to the PLC, according to PLCTYPE constants

pyads.pyads_ex.get_value_from_ctype_data(read_data: Any | None, plc_type: Type) → Any[source]

Convert ctypes data object to a regular value based on the PLCTYPE_* property.

Typical usage is:

obj = my_plc_type.from_buffer(my_buffer)
value = get_value_from_ctype_data(obj, my_plc_type)

Parameters:

read_data – ctypes._CData object
plc_type – pyads.PLCTYPE_* constant (i.e. a ctypes-like type)

pyads.pyads_ex.router_function(fn: Callable) → Callable[source]

Raise a runtime error if on Win32 systems.

Decorator.

Decorator for functions that interact with the router for the Linux implementation of the ADS library.

Unlike the Windows implementation which uses a separate router daemon, the Linux library manages AMS routing in-process. As such, routing must be configured programmatically via. the provided API. These endpoints are invalid on Win32 systems, so an exception will be raised.

pyads.pyads_ex.send_raw_udp_message(ip_address: str, message: bytes, expected_return_length: int) → Tuple[bytes, Tuple[str, int]][source]

Send a raw UDP message to the PLC and return the response.

Parameters:

ip_address (str) – ip address of the PLC
message (bytes) – the message to send to the PLC
expected_return_length (int) – number of bytes to expect in response

Return type:

Tuple[bytes, Tuple[str, int]]

Returns:

A tuple containing the response and a tuple containing the IP address and port of the sending socket

pyads.pyads_ex.type_is_string(plc_type: Type) → bool[source]: Return true if the given class is a string type.

pyads.pyads_ex.type_is_wstring(plc_type: Type) → bool[source]: Return True if the given class is a WSTRING type.

pyads.structs module

Structs for the work with ADS API.

author:: Stefan Lehmann <stlm@posteo.de>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2018-06-11 18:15:53

class pyads.structs.AdsVersion(stAdsVersion: SAdsVersion)[source]

Bases: object

Contains version number, revision number, build number of the ADS-DLL.

Variables:

version (int) – version number
revision (int) – revision number
build (int) – build number

Create new AdsVersion object.

Parameters:: stAdsVersion (pyads.constants.SAdsVersion) – ctypes structure with the version info

class pyads.structs.AmsAddr(netid: str = None, port: int = None)[source]

Bases: object

Wrapper for SAmsAddr-structure to address an ADS device.

Variables:: _ams_addr – ctypes-structure SAmsAddr

Create a new AmsAddr object by a given netid and port.

Parameters:

netid – NetId of an ADS device
port – port of an ADS device

amsAddrStruct() → SAmsAddr[source]: Return the c-types structure SAmsAddr.

netIdStruct() → SAmsNetId[source]: Return the c-types structure SAmsNetId.

property netid: str

Netid of the AmsAddress.

The Netid is always returned as a String. If the NetId is set it can be passed as a String or as a SAmsNetId struct.

property port: int: Port of the AmsAddress object.

setAdr(adrString: str) → None[source]

Set the AMS-address according to the given IP-address.

Parameters:: adrString (string) – ip-address of an ADS device

toString() → str[source]

Textual representation of the AMS address.

Return type:: string
Returns:: textual representation of the AMS address

class pyads.structs.NotificationAttrib(length: int, trans_mode: int = 4, max_delay: float = 0.0001, cycle_time: float = 0.0001)[source]

Bases: object

Notification Attribute.

Create a new NotificationAttrib object.

Parameters:

length (int) – length of the data
trans_mode (int) – transmission mode
max_delay (float) – maximum delay in ms
cycle_time (float) – cycle time in ms

property cycle_time: int: Notification cycle time in ms for cycle transmission mode.

property length: int: Notification data length.

property max_delay: None: Maximum allowed delay between notifications in ms.

notificationAttribStruct() → SAdsNotificationAttrib[source]: Return the raw struct.

property trans_mode: int

Mode of transmission.

This can be one of the following:

ADSTRANS_NOTRANS
ADSTRANS_CLIENTCYCLE
ADSTRANS_CLIENT1REQ
ADSTRANS_SERVERCYCLE
ADSTRANS_SERVERONCHA

class pyads.structs.SAdsNotificationAttrib[source]

Bases: Structure

C structure representation of AdsNotificationAttrib.

AttribUnion: Structure/Union member

cbLength: Structure/Union member

dwChangeFilter: Structure/Union member

nCycleTime: Structure/Union member

nMaxDelay: Structure/Union member

nTransMode: Structure/Union member

class pyads.structs.SAdsNotificationHeader[source]

Bases: Structure

C structure representation of AdsNotificationHeader.

Variables:

hNotification – notification handle
nTimeStamp – time stamp in FILETIME format
cbSampleSize – number of data bytes
data – variable-length data field, get via ctypes.addressof + offset

cbSampleSize: Structure/Union member

data: Structure/Union member

hNotification: Structure/Union member

nTimeStamp: Structure/Union member

class pyads.structs.SAdsSumRequest[source]

Bases: Structure

ADS sum request structure.

Variables:

iGroup – indexGroup of request
iOffs – indexOffset of request
size – size of request

iGroup: Structure/Union member

iOffset: Structure/Union member

size: Structure/Union member

class pyads.structs.SAdsSymbolEntry[source]

Bases: Structure

ADS symbol information.

Variables:

entryLength – length of complete symbol entry
iGroup – indexGroup of symbol: input, output etc.
iOffs – indexOffset of symbol
size – size of symbol (in bytes, 0=bit)
dataType – adsDataType of symbol
flags – symbol flags
nameLength – length of symbol name
typeLength – length of type name
commentLength – length of comment

A complete example could be:

value: 57172            # Current value
info.entryLength: 88    # Total storage space for this symbol
info.iGroup: 16448      # Group index
info.iOffs: 385000      # Offset index inside group
info.size: 2            # Number of bytes needed for the value
info.dataType: 18       # Symbol type, in this case
                          constants.ADST_UINT16 (18)
info.flags: 8           # TwinCAT byte flags
info.nameLength: 11     # Number of characters in the name
info.typeLength: 4      # Number of characters in the PLC string
                          representation of the type
info.commentLength: 20  # Number of characters in the comment
info.stringBuffer: <pyads.structs.c_ubyte_Array_768 object>
                        # Concatenation of all string info
bytes(info.stringBuffer): b'GVL.counterUINTCounter (in '
                          'pulses)'
bytes(info.stringBuffer).encode(): "GVL.counter UINT Counter (in
                                    pulses)"

info.name: "GVL.counter"    # The name section from the buffer
info.symbol_type: "UINT"    # The symbol_type section from the
                              buffer
info.comment: " Counter (in pulses)"  # The comment (if any)

property comment: str: User-defined comment.

commentLength: Structure/Union member

dataType: Structure/Union member

entryLength: Structure/Union member

flags: Structure/Union member

iGroup: Structure/Union member

iOffs: Structure/Union member

property name: str: The symbol name.

nameLength: Structure/Union member

size: Structure/Union member

stringBuffer: Structure/Union member

property symbol_type: str: The qualified type name, including the namespace.

typeLength: Structure/Union member

class pyads.structs.SAdsSymbolUploadInfo[source]

Bases: Structure

C structure representation of AdsSymbolUploadInfo.

nSymSize: Structure/Union member

nSymbols: Structure/Union member

class pyads.structs.SAdsVersion[source]

Bases: Structure

Struct containing ADS version information.

build: Structure/Union member

revision: Structure/Union member

version: Structure/Union member

class pyads.structs.SAmsAddr[source]

Bases: Structure

Struct containing the netId and port of an ADS device.

netId: Structure/Union member

port: Structure/Union member

class pyads.structs.SAmsNetId[source]

Bases: Structure

Struct with array of 6 bytes used to describe a net id.

b: Structure/Union member

pyads.symbol module

Define the Symbol class

Separate file because it depends on many other files, so we try to simplify the circular dependencies.

author:: Roberto Roos
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2020-11-16

Bases: object

Object that points to an ADS variable

Contains index group, index offset, name, symbol type, comment of ADS symbol. Also remembers a reference to a Connection to be able to read/write directly.

The virtual property value can be used to read from and write to the symbol.

Variables:

index_group – Index group of symbol
index_offset – Index offset of symbol
name – Name of symbol
symbol_type – String representation of symbol type (PLC-style, e.g. “LREAL”)
plc_type – ctypes type of variable (from constants.PLCTYPE_*)
comment – Comment of symbol
value – Buffered value, i.e. the most recently read or written value for this symbol

Create AdsSymbol 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.

symbol_type should be a type constant like pyads.PLCTYPE_*. Alternatively, it can be a string representation a PLC type (e.g. ‘LREAL’).

Parameters:

plc – Connection instance
name
index_group
index_offset
symbol_type – PLC variable type (e.g. pyads.PLCTYPE_DINT)
comment
auto_update – Create notification to update buffer (same as set_auto_update(True))
structure_def (Optional["StructureDef"]) – 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’}
array_size (Optional[int]) – size of array if reading array of structure, defaults to 1

Expected input example for structure_def:

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))

add_device_notification(callback: Callable[[Any, Any], None], attr: NotificationAttrib | None = None, user_handle: int | None = None) → Tuple[int, int] | None[source]

Add on-change callback to symbol.

See Connection.add_device_notification(…).

When attr is omitted, the default will be used.

The notification handles are returned but also stored locally. When this symbol is destructed any notifications will be freed up automatically.

property auto_update: Any: Return True if auto_update is enabled for this symbol.

clear_device_notifications() → None[source]: Remove all registered notifications

del_device_notification(handles: Tuple[int, int]) → None[source]: Remove a single device notification by handles

Get PLCTYPE_* from PLC name string

If PLC name could not be mapped, return None. This is done on purpose to prevent a program from crashing when an unusable symbol is found. Instead, exceptions will be thrown when this unmapped symbol is read/written.

property is_structure: bool

Return True if the symbol object represents a structure.

This is the case if a structure_def has been passed during initialization.

read() → Any[source]

Read the current value of this symbol.

The new read value is also saved in the buffer.

property value: Any: Return the current value of the symbol.

write(new_value: Any | None = None) → None[source]

Write a new value or the buffered value to the symbol.

When a new value was written, the buffer is updated.

:param new_value Value to be written to symbol (if None,: the buffered value is send instead)

pyads.testserver module

The testserver package of pyads.

author:: Roberto Roos
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2021-04-09

pyads.testserver.testserver

Extended ADS TCP/IP server implementation.

Extended ADS TCP/IP server implementation to allow for functional testing of the ADS protocol without connection to a physical device.

Consists of a server thread which will listen for connections and delegate each new connection to a separate client thread, allowing for multiple clients to connect at once.

Each client connection thread listens for incoming data, and delegates parsing and response construction to the handler. A handler function is injectable at server level by specifying the handler kwarg in the server constructor.

author:: David Browne <davidabrowne@gmail.com>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2018-06-11 18:15:53

class pyads.testserver.testserver.AdsClientConnection(handler: AbstractHandler, client: socket, address: str, server: AdsTestServer, *args: Any, **kwargs: Any)[source]

Bases: Thread

Connection thread to an ADS client.

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

close() → None[source]: Close the client connection.

static construct_request(request_bytes: bytes) → AmsPacket[source]

Unpack an AMS packet from binary data.

Parameters:: request_bytes (bytes) – The raw request data
Rtype AmsPacket:
Returns:: AmsPacket with fields populated from the binary data

static construct_response(response_data: AmsResponseData, request: AmsPacket) → bytes[source]

Construct binary AMS response to return to the client.

Parameters:

response_data (AmsResponseData) – Data to include in the response
request (AmsPacket) – The originating request for the response

run() → None[source]: Listen for data on client connection and delegate requests.

stop() → None[source]: Stop the client thread.

class pyads.testserver.testserver.AdsTestServer(handler: AbstractHandler = None, ip_address: str = '127.0.0.1', port: int = 48898, logging: bool = True, *args: Any, **kwargs: Any)[source]

Bases: Thread

Simple ADS testing server.

Variables:

handler (function) – Request handler (see default_handler for example)
ip_address (str) – Host address for server. Defaults to ‘127.0.0.1’
port (int) – Host port to listen on, defaults to 48898

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

close() → None[source]: Close the server thread.

run() → None[source]: Listen for incoming connections from clients.

stop() → None[source]: Close client connections and stop main server loop.

pyads.testserver.handler

Abstract handler module for testserver.

author:: David Browne <davidabrowne@gmail.com>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2016-09-13

class pyads.testserver.handler.AbstractHandler[source]

Bases: object

Abstract Handler class to provide a base class for handling requests.

handle_request(request: AmsPacket) → AmsResponseData[source]

Handle incoming requests.

Parameters:: request (AmsPacket) – The request data received from the client
Return type:: AmsResponseData
Returns:: Data needed to construct the AMS response packet

class pyads.testserver.handler.AmsHeader(target_net_id, target_port, source_net_id, source_port, command_id, state_flags, length, error_code, invoke_id, data)

Bases: tuple

Create new instance of AmsHeader(target_net_id, target_port, source_net_id, source_port, command_id, state_flags, length, error_code, invoke_id, data)

command_id: Alias for field number 4

data: Alias for field number 9

error_code: Alias for field number 7

invoke_id: Alias for field number 8

length: Alias for field number 6

source_net_id: Alias for field number 2

source_port: Alias for field number 3

state_flags: Alias for field number 5

target_net_id: Alias for field number 0

target_port: Alias for field number 1

class pyads.testserver.handler.AmsPacket(tcp_header, ams_header)

Bases: tuple

Create new instance of AmsPacket(tcp_header, ams_header)

ams_header: Alias for field number 1

tcp_header: Alias for field number 0

class pyads.testserver.handler.AmsResponseData(state_flags, error_code, data)

Bases: tuple

Create new instance of AmsResponseData(state_flags, error_code, data)

data: Alias for field number 2

error_code: Alias for field number 1

state_flags: Alias for field number 0

class pyads.testserver.handler.AmsTcpHeader(length)

Bases: tuple

Create new instance of AmsTcpHeader(length,)

length: Alias for field number 0

pyads.testserver.basic_handler

Basic handler module for testserver.

author:: Stefan Lehmann <stlm@posteo.de>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2017-09-15

class pyads.testserver.basic_handler.BasicHandler[source]

Bases: AbstractHandler

Basic request handler.

Basic request handler to print the request data and return some default values.

handle_request(request: AmsPacket) → AmsResponseData[source]: Handle incoming requests and send a response.

pyads.testserver.advanced_handler

Advanced handler module for testserver.

author:: Stefan Lehmann <stlm@posteo.de>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2017-09-15

class pyads.testserver.advanced_handler.AdvancedHandler[source]

Bases: AbstractHandler

The advanced handler allows to store and restore data.

The advanced handler allows to store and restore data via read, write and read_write functions. There is a storage area for each symbol. The purpose of this handler to test read/write access and test basic interaction. Variables can be read/write through indices, name and handle.

An error will be thrown when an attempt is made to read from a non-existent variable. You can either: i) write the variable first (it is implicitly created) or ii) create the variable yourself and place it in the handler. Note that the variable type cannot be set correctly in the implicit creation! (It will default to UINT16.) Use explicit creation if a non-default type is important.

add_variable(var: PLCVariable) → None[source]: Add a new variable.

get_variable_by_handle(handle: int) → PLCVariable[source]: Get PLC variable by handle, throw error when not found

get_variable_by_indices(index_group: int, index_offset: int) → PLCVariable[source]: Get PLC variable by handle, throw error when not found

get_variable_by_name(name: str) → PLCVariable[source]: Get variable by name, throw error if not found

get_variable_by_notification_handle(handle: int) → PLCVariable[source]: Get variable by a notification handle, throw error if not found

handle_request(request: AmsPacket) → AmsResponseData[source]: Handle incoming requests and create a response.

reset() → None[source]: Clear saved variables in handler

class pyads.testserver.advanced_handler.PLCVariable(name: str, value: int | float | bytes, ads_type: int, symbol_type: str, index_group: int | None = None, index_offset: int | None = None)[source]

Bases: object

Storage item for named data.

Also include variable type so it can be retrieved later. This basically mirrors SAdsSymbolEntry or AdsSymbol, however we want to avoid using those directly since they are test subjects.

Handle and indices are set by default (to random but safe values)

Parameters:

name (str) – variable name
value (bytes) – variable value as bytes
ads_type (int) – constants.ADST_*
symbol_type (str) – PLC-style name of type
index_group (Optional[int]) – set index_group manually
index_offset (Optional[int]) – set index_offset manually

INDEX_GROUP = 12345

INDEX_OFFSET_BASE = 10000

get_packed_info() → bytes[source]: Get bytes array of symbol info

handle_count = 10000

notification_count = 10

register_notification() → int[source]: Register a new notification.

property size: int: Return size of value.

unregister_notification(handle: int = None)[source]

Unregister a notification.

Parameters:: handle – Set to None (default) to unregister all notifications

write(value: bytes, request: AmsPacket = None)[source]: Update the variable value, respecting notifications

pyads.utils module

Utility functions.

author:: Stefan Lehmann <stlm@posteo.de>
license:: MIT, see license file or https://opensource.org/licenses/MIT
created on:: 2018-06-11 18:15:53

pyads.utils.decode_ads(message: bytes) → str[source]

Decode a string that in encoded in the format used by ADS.

From Beckhoff documentation: ‘A STRING constant is a string enclosed by single quotation marks. The characters are encoded according to the Windows 1252 character set. As a subset of Windows-1252, the character set of ISO/IEC 8859-1 is supported.’

pyads.utils.deprecated(message: str | None = None) → Callable[source]

Decorator for deprecated functions.

Shows a deprecation warning with the given message if the decorated function is called.

pyads.utils.find_wstring_null_terminator(data: bytearray) → int | None[source]

Find null-terminator in WSTRING (UTF-16) data.

Returns:: None if no null-terminator was found, else the index of the null-terminator

pyads.utils.get_num_of_chars(symbol_type_str: <property object at 0x7fa129ba8400>) → int[source]

Gets the number of characters in a Beckhoff string using the symbol type str.

TODO: Find this information some other way without string manipulation of the symbol type?

Args:: symbol_type_str (SAdsSymbolEntry.symbol_type): Symbol type of a string, wstring, or a string array
Returns:: num_characters (int): The number of characters im the string

pyads.utils.platform_is_beckhoff_rt_linux() → bool[source]: Workaround to identify Beckhoff RT-Linux as a FreeBSD-based solution instead of Linux.

pyads.utils.platform_is_freebsd() → bool[source]: Return True if current platform is FreeBSD.

pyads.utils.platform_is_linux() → bool[source]: Return True if current platform is Linux or Mac OS.

pyads.utils.platform_is_windows() → bool[source]: Return True if current platform is Windows.

Module contents

The pyads package.

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