pyads package

The submodules of the Pyads package are listed below.

pyads.ads module

Pythonic ADS 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.ads.add_route(adr: Optional[Union[str, pyads.structs.AmsAddr]], 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: Optional[str] = None, added_net_id: Optional[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: Union[Dict[str, Any], List[Dict[str, Any]]], structure_def: Tuple[Union[Tuple[str, Type, int], Tuple[str, Type, int, Optional[int]]], ...]) 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: pyads.structs.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[Union[Tuple[str, Type, int], Tuple[str, Type, int, Optional[int]]], ...], array_size: int = 1) Union[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() Optional[pyads.structs.AmsAddr][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: Union[str, pyads.structs.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[Union[Tuple[str, Type, int], Tuple[str, Type, int, Optional[int]]], ...]) 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[_ctypes.Array][source]

Return an array with n boolean values.

pyads.constants.PLCTYPE_ARR_DINT(n: int) Type[_ctypes.Array][source]

Return an array with n int32 values.

pyads.constants.PLCTYPE_ARR_INT(n: int) Type[_ctypes.Array][source]

Return an array with n int16 values.

pyads.constants.PLCTYPE_ARR_LREAL(n: int) Type[_ctypes.Array][source]

Return an array with n double values.

pyads.constants.PLCTYPE_ARR_REAL(n: int) Type[_ctypes.Array][source]

Return an array with n float values.

pyads.constants.PLCTYPE_ARR_SHORT(n: int) Type[_ctypes.Array][source]

Return an array with n short values.

pyads.constants.PLCTYPE_ARR_SINT(n: int) Type[_ctypes.Array][source]

Return an array with n int8 values.

pyads.constants.PLCTYPE_ARR_UDINT(n: int) Type[_ctypes.Array][source]

Return an array with n uint32 values.

pyads.constants.PLCTYPE_ARR_UINT(n: int) Type[_ctypes.Array][source]

Return an array with n uint16 values.

pyads.constants.PLCTYPE_ARR_USHORT(n: int) Type[_ctypes.Array][source]

Return an array with n ushort values.

pyads.constants.PLCTYPE_ARR_USINT(n: int) Type[_ctypes.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>

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: Optional[str] = None, ams_net_port: Optional[int] = None, ip_address: Optional[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: Union[str, Tuple[int, int]], attr: pyads.structs.NotificationAttrib, callback: Callable, user_handle: Optional[int] = None) Optional[Tuple[int, int]][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
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[pyads.symbol.AdsSymbol][source]

Read all symbols from an ADS-device.

Returns

List of AdsSymbols

get_handle(data_name: str) Optional[int][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() Optional[pyads.structs.AmsAddr][source]

Return the local AMS-address and the port number.

Return type

AmsAddr

get_symbol(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) pyads.symbol.AdsSymbol[source]

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: Optional[Type] = 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: Optional[Type], timestamp_as_filetime: bool = False) Tuple[int, Union[datetime.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(index_group: int, index_offset: int, plc_datatype: Type[PLCDataType], return_ctypes: bool = False, check_length: bool = True) Any[source]

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_by_name(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[source]

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() Optional[Tuple[str, pyads.structs.AdsVersion]][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: Optional[Dict[str, Tuple[Union[Tuple[str, Type, int], Tuple[str, Type, int, Optional[int]]], ...]]] = 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() Optional[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.

Return type

(int, int)

Returns

adsState, deviceState

read_structure_by_name(data_name: str, structure_def: Tuple[Union[Tuple[str, Type, int], Tuple[str, Type, int, Optional[int]]], ...], array_size: Optional[int] = 1, structure_size: Optional[int] = None, handle: Optional[int] = None) Optional[Union[Dict[str, Any], List[Dict[str, Any]]]][source]

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_write(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[source]

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.

write(index_group: int, index_offset: int, value: Any, plc_datatype: Type[PLCDataType]) None[source]

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

write_by_name(data_name: str, value: Any, plc_datatype: Optional[Type[PLCDataType]] = None, handle: Optional[int] = None, cache_symbol_info: bool = True) None[source]

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: Optional[Dict[str, Tuple[Union[Tuple[str, Type, int], Tuple[str, Type, int, Optional[int]]], ...]]] = 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_structure_by_name(data_name: str, value: Union[Dict[str, Any], List[Dict[str, Any]]], structure_def: Tuple[Union[Tuple[str, Type, int], Tuple[str, Type, int, Optional[int]]], ...], array_size: Optional[int] = 1, structure_size: Optional[int] = None, handle: Optional[int] = None) None[source]

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

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: Optional[int] = None, text: Optional[str] = None)[source]

Bases: Exception

Error class for errors related to ADS communication.

pyads.pyads_ex.adsAddRoute(net_id: pyads.structs.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: Optional[str] = None, added_net_id: Optional[str] = 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: pyads.structs.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: pyads.structs.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) pyads.structs.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: pyads.structs.AmsAddr, data_name: str) pyads.structs.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: pyads.structs.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: pyads.structs.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: pyads.structs.AmsAddr, data_names: List[str], data_symbols: Dict[str, pyads.structs.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: pyads.structs.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: pyads.structs.AmsAddr, data_names_and_values: Dict[str, Any], data_symbols: Dict[str, pyads.structs.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: pyads.structs.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: pyads.structs.AmsAddr, data: Union[str, Tuple[int, int]], pNoteAttrib: pyads.structs.NotificationAttrib, callback: Callable, user_handle: Optional[int] = 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: pyads.structs.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: pyads.structs.AmsAddr, data_name: str, data_type: Type, return_ctypes: bool = False, handle: Optional[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: pyads.structs.AmsAddr) Tuple[str, pyads.structs.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: pyads.structs.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: pyads.structs.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
Return type

(int, int)

Returns

ads_state, device_state

pyads.pyads_ex.adsSyncReadWriteReqEx2(port: int, address: pyads.structs.AmsAddr, index_group: int, index_offset: int, read_data_type: Optional[Type], value: Any, write_data_type: Optional[Type], 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: pyads.structs.AmsAddr, data_name: str, value: Any, data_type: Type, handle: Optional[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: pyads.structs.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: pyads.structs.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: Optional[Any], 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: pyads.structs.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: Optional[str] = None, port: Optional[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() pyads.structs.SAmsAddr[source]

Return the c-types structure SAmsAddr.

netIdStruct() pyads.structs.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() pyads.structs.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: _ctypes.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: _ctypes.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: _ctypes.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: _ctypes.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: _ctypes.Structure

C structure representation of AdsSymbolUploadInfo.

nSymSize

Structure/Union member

nSymbols

Structure/Union member

class pyads.structs.SAdsVersion[source]

Bases: _ctypes.Structure

Struct containing ADS version information.

build

Structure/Union member

revision

Structure/Union member

version

Structure/Union member

class pyads.structs.SAmsAddr[source]

Bases: _ctypes.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: _ctypes.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

class pyads.symbol.AdsSymbol(plc: Connection, name: Optional[str] = None, index_group: Optional[int] = None, index_offset: Optional[int] = None, symbol_type: Optional[Union[Type[PLCDataType], str]] = None, comment: Optional[str] = None, auto_update: bool = False, structure_def: Optional[StructureDef] = None, array_size: Optional[int] = 1)[source]

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: Optional[pyads.structs.NotificationAttrib] = None, user_handle: Optional[int] = None) Optional[Tuple[int, int]][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

static get_type_from_str(type_str: str) Optional[Type[Union[_ctypes.Array, ctypes.c_bool, ctypes.c_ubyte, ctypes.c_uint, ctypes.c_int, ctypes.c_short, ctypes.c_double, ctypes.c_float, ctypes.c_byte, ctypes.c_char, pyads.constants.PLCTYPE_WSTRING, ctypes.c_ushort, ctypes.c_long, ctypes.c_ulong]]][source]

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: Optional[Any] = 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: pyads.testserver.handler.AbstractHandler, client: socket.socket, address: str, server: pyads.testserver.testserver.AdsTestServer, *args: Any, **kwargs: Any)[source]

Bases: threading.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 the argument tuple 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) pyads.testserver.handler.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: pyads.testserver.handler.AmsResponseData, request: pyads.testserver.handler.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: Optional[pyads.testserver.handler.AbstractHandler] = None, ip_address: str = '127.0.0.1', port: int = 48898, logging: bool = True, *args: Any, **kwargs: Any)[source]

Bases: threading.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 the argument tuple 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: pyads.testserver.handler.AmsPacket) pyads.testserver.handler.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)

property command_id

Alias for field number 4

property data

Alias for field number 9

property error_code

Alias for field number 7

property invoke_id

Alias for field number 8

property length

Alias for field number 6

property source_net_id

Alias for field number 2

property source_port

Alias for field number 3

property state_flags

Alias for field number 5

property target_net_id

Alias for field number 0

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

property ams_header

Alias for field number 1

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

property data

Alias for field number 2

property error_code

Alias for field number 1

property state_flags

Alias for field number 0

class pyads.testserver.handler.AmsTcpHeader(length)

Bases: tuple

Create new instance of AmsTcpHeader(length,)

property 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: pyads.testserver.handler.AbstractHandler

Basic request handler.

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

handle_request(request: pyads.testserver.handler.AmsPacket) pyads.testserver.handler.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: pyads.testserver.handler.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: pyads.testserver.advanced_handler.PLCVariable) None[source]

Add a new variable.

get_variable_by_handle(handle: int) pyads.testserver.advanced_handler.PLCVariable[source]

Get PLC variable by handle, throw error when not found

get_variable_by_indices(index_group: int, index_offset: int) pyads.testserver.advanced_handler.PLCVariable[source]

Get PLC variable by handle, throw error when not found

get_variable_by_name(name: str) pyads.testserver.advanced_handler.PLCVariable[source]

Get variable by name, throw error if not found

get_variable_by_notification_handle(handle: int) pyads.testserver.advanced_handler.PLCVariable[source]

Get variable by a notification handle, throw error if not found

handle_request(request: pyads.testserver.handler.AmsPacket) pyads.testserver.handler.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: Union[int, float, bytes], ads_type: int, symbol_type: str, index_group: Optional[int] = None, index_offset: Optional[int] = 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: Optional[int] = None)[source]

Unregister a notification.

Parameters

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

write(value: bytes, request: Optional[pyads.testserver.handler.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: Optional[str] = None) Callable[source]

Decorator for deprecated functions.

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

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>

license

MIT, see license file or https://opensource.org/licenses/MIT

created on

2018-06-11 18:15:53