Welcome to pyads’s documentation!

This is a Python wrapper for TwinCATs ADS library. It aims to provide a pythonic way to communicate with TwinCAT devices by using the Python programming language. pyads uses the C API provided by TcAdsDll.dll on Windows and adslib.so on Linux. The Linux library is included in this package.

The documentation for the ADS API is available on infosys.beckhoff.com.

Installation

From PyPi:

pip install pyads

From conda-forge:

conda install pyads

From source:

git clone https://github.com/MrLeeh/pyads.git --recursive
cd pyads
python setup.py install

Installation on Linux

For Linux pyads uses the ADS library adslib.so which needs to be compiled from source if you use a source package. This should not be an issue, however if you should encounter any problems with the adslib.so please contact me.

Installation on Windows

On Windows pyads uses the TcADSDll.dll which is provided when you install Beckhoffs TwinCAT on your machine. Make sure that it is accessible and installed in your PATH.

Testing your installation

You can test your installation by simply popping up a python console and importing the pyads module. If no errors occur everything is fine and you can carry on.

>>> import pyads

If you get an OSError saying that the adslib.so could not be found there probably went something wrong with the build process of the shared library. In this case you can create the adslib.so manually by doing the following:

cd adslib
make
sudo make install

This compiles and places the adslib.so in your /usr/lib/ directory.

Quickstart

Important

You need to create routes on your client pc and on your target plc via TwinCAT before any communication can take place. If you are on Windows you can use the TwinCAT router. For Linux systems the route is created automatically on the client-side. For the target-side you can use add_route_to_plc().

>>> import pyads

>>> # create some constants for connection
>>> CLIENT_NETID = "192.168.1.10.1.1"
>>> CLIENT_IP = "192.168.1.10"
>>> TARGET_IP = "192.168.1.11"
>>> TARGET_USERNAME = "Administrator"
>>> TARGET_PASSWORD = "1"
>>> ROUTE_NAME = "route-to-my-plc"

>>> # add a new route to the target plc
>>> pyads.add_route_to_plc(
>>>     CLIENT_NETID, CLIENT_IP, TARGET_IP, TARGET_USERNAME, TARGET_PASSWORD,
>>>     route_name=ROUTE_NAME
>>> )

>>> # connect to plc and open connection
>>> # route is added automatically to client on Linux, on Windows use the TwinCAT router
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_SPS1)
>>> plc.open()

>>> # check the connection state
>>> plc.read_state()
(0, 5)

>>> # read int value by name
>>> i = plc.read_by_name("GVL.int_val")

>>> # write int value by name
>>> plc.write_by_name("GVL.real_val", 42.0)

>>> # create a symbol that automatically updates to the plc value
>>> real_val = plc.get_symbol("GVL.real_val", auto_update=True)
>>> print(real_val.value)
42.0
>>> real_val.value = 5.0
>>> print(plc.read_by_name("GVL.real_val"))
5.0

>>> # close connection
>>> plc.close()

Documentation

Routing

ADS uses its own address system named AmsNetId to identify devices. The assignment of a devices to an AmsNetId happens via routing. Routing is handled differently on Windows and Linux and is explained for both operating systems in the sections below.

To identify each side of a route we will use the terms client and target. The client is your computer where pyads runs on. The target is you plc or remote computer which you want to connect to.

Creating routes on Windows

On Windows you don’t need to manually add the routes with pyads but instead you use the TwinCAT Router UI ( TcSystemManager) which comes with the TwinCAT installation. Have a look at the TwinCAT documentation on infosys.beckhoff.com for further details.

Creating routes on Linux

To create a new route on Linux you can simply use the Connection class. It connects to the target and creates a route to it on your client.

>>> import pyads
>>> remote_ip = '192.168.0.100'
>>> remote_ads = '5.12.82.20.1.1'
>>> with pyads.Connection(remote_ads, pyads.PORT_TC3PLC1, remote_ip) as plc:
>>>     plc.read_by_name('.TAG_NAME', pyads.PLCTYPE_INT)

Note

You still need to create a route from the target to the client. You can do this manually on your target or you can use the function add_route_to_plc() as explained below!

Get the AMS address of the local machine. This may need to be added to the routing table of the remote machine.

Note

On Linux machines at least one route must be added before the call to get_local_address() will function properly.

Optionally, a local AmsNetId can be manually set before adding a route. Set this to match the expected AMS ID in the remote machine’s routing table.

>> > import pyads
>> > pyads.open_port()
>> > pyads.set_local_address('1.2.3.4.1.1')
>> > pyads.close_port()

Adding routes to a target

ADS requires you to create a route in the routing tables of both your client and your target. How you add a route to your client is handled in the section above. To create a route on your target you can either use TwinCAT or you can make use of the convenience function add_route_to_plc().

Here is an example of adding a route to a target (e.g. remote plc) to allow connections to a PC with the Hostname “MyPC”

Warning

You need to open a port and set a local netid with set_local_address() before you can use add_route_to_plc().

>>> import pyads
>>> SENDER_AMS = '1.2.3.4.1.1'
>>> PLC_IP = '192.168.0.100'
>>> PLC_USERNAME = 'plc_username'
>>> PLC_PASSWORD = 'plc_password'
>>> ROUTE_NAME = 'RouteToMyPC'
>>> HOSTNAME = 'MyPC'  # or IP
>>>
>>> pyads.open_port()
>>> pyads.set_local_address(SENDER_AMS)
>>> pyads.add_route_to_plc(SENDER_AMS, HOSTNAME, PLC_IP, PLC_USERNAME, PLC_PASSWORD, route_name=ROUTE_NAME)
>>> pyads.close_port()

Note

When adding the route in TwinCAT make sure to deactivate the unidirectional option.

Connections

Important

Before starting a connection to a target make sure you created proper routes on the client and the target like described in the Routing chapter.

Connect to a remote device

>>> import pyads
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
>>> plc.open()
>>> plc.close()

The connection will be closed automatically if the object runs out of scope, making Connection.close() optional.

A context notation (using with:) can be used to open a connection:

>>> import pyads
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
>>> with plc:
>>>     # ...

The context manager will make sure the connection is closed, either when the with clause runs out, or an uncaught error is thrown.

Read and write by name

Values

Reading and writing values from/to variables on the target can be done with Connection.read_by_name() and Connection.write_by_name(). Passing the plc_datatype is optional for both methods. If plc_datatype is None the datatype will be queried from the target on the first call and cached inside the Connection object. You can disable symbol-caching by setting the parameter cache_symbol_info to False.

Warning

Querying the datatype only works for basic datatypes. For structs, lists and lists of structs you need provide proper definitions of the datatype and use Connection.read_structure_by_name() or Connection.read_list_by_name().

Examples:

 >>> import pyads
 >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1):
 >>> plc.open()
 >>>
 >>> plc.read_by_name('GVL.bool_value')  # datatype will be queried and cached
 True
 >>> plc.read_by_name('GVL.bool_value')  # cached datatype will be used
 True
 >>> plc.read_by_name('GVL.bool_value', cache_symbol_info=False)  # datatype will not be cached and queried on each call
 True
 >>> plc.read_by_name('GVL.int_value', pyads.PLCTYPE_INT)  # datatype is provided and will not be queried
 0
 >>> plc.write_by_name('GVL.int_value', 10)  # write to target
 >>> plc.read_by_name('GVL.int_value')
 10

>>> plc.close()

If the name could not be found an Exception containing the error message and ADS Error number is raised.

>>> plc.read_by_name('GVL.wrong_name', pyads.PLCTYPE_BOOL)
ADSError: ADSError: symbol not found (1808)

For reading strings the maximum buffer length is 1024.

>>> plc.read_by_name('GVL.sample_string', pyads.PLCTYPE_STRING)
'Hello World'
>>> plc.write_by_name('GVL.sample_string', 'abc', pyads.PLCTYPE_STRING)
>>> plc.read_by_name('GVL.sample_string', pyads.PLCTYPE_STRING)
'abc'

Arrays

You can also read/write arrays. For this you simply need to multiply the datatype by the number of elements in the array or structure you want to read/write.

>>> plc.write_by_name('GVL.sample_array', [1, 2, 3], pyads.PLCTYPE_INT * 3)
>>> plc.read_by_name('GVL.sample_array', pyads.PLCTYPE_INT * 3)
[1, 2, 3]

>>> plc.write_by_name('GVL.sample_array[0]', 5, pyads.PLCTYPE_INT)
>>> plc.read_by_name('GVL.sample_array[0]', pyads.PLCTYPE_INT)
5

Structures of the same datatype

TwinCAT declaration:

TYPE sample_structure :
STRUCT
    rVar : LREAL;
    rVar2 : LREAL;
    rVar3 : LREAL;
    rVar4 : ARRAY [1..3] OF LREAL;
END_STRUCT
END_TYPE

Python code:

>>> plc.write_by_name('GVL.sample_structure',
                      [11.1, 22.2, 33.3, 44.4, 55.5, 66.6],
                      pyads.PLCTYPE_LREAL * 6)
>>> plc.read_by_name('GVL.sample_structure', pyads.PLCTYPE_LREAL * 6)
[11.1, 22.2, 33.3, 44.4, 55.5, 66.6]

>>> plc.write_by_name('GVL.sample_structure.rVar2', 1234.5, pyads.PLCTYPE_LREAL)
>>> plc.read_by_name('GVL.sample_structure.rVar2', pyads.PLCTYPE_LREAL)
1234.5

Structures with multiple datatypes

The structure in the PLC must be defined with `{attribute ‘pack_mode’ := ‘1’}.

TwinCAT declaration:

{attribute 'pack mode' := '1'}
TYPE sample_structure :
STRUCT
    rVar : LREAL;
    rVar2 : REAL;
    iVar : INT;
    iVar2 : ARRAY [1..3] OF DINT;
    sVar : STRING;
END_STRUCT
END_TYPE

Python code:

First declare a tuple which defines the PLC structure. This should match the order as declared in the PLC. Information is passed and returned using the OrderedDict type.

>>> structure_def = (
...    ('rVar', pyads.PLCTYPE_LREAL, 1),
...    ('rVar2', pyads.PLCTYPE_REAL, 1),
...    ('iVar', pyads.PLCTYPE_INT, 1),
...    ('iVar2', pyads.PLCTYPE_DINT, 3),
...    ('sVar', pyads.PLCTYPE_STRING, 1)
... )

>>> vars_to_write = OrderedDict([
...     ('rVar', 11.1),
...     ('rar2', 22.2),
...     ('iVar', 3),
...     ('iVar2', [4, 44, 444]),
...     ('sVar', 'abc')]
... )

>>> plc.write_structure_by_name('global.sample_structure', vars_to_write, structure_def)
>>> plc.read_structure_by_name('global.sample_structure', structure_def)
OrderedDict([('rVar', 11.1), ('rVar2', 22.2), ('iVar', 3), ('iVar2', [4, 44, 444]), ('sVar', 'abc')])

Read and write by handle

When reading and writing by name, internally pyads is acquiring a handle from the PLC, reading/writing the value using that handle, before releasing the handle. A handle is just a unique identifier that the PLC associates to an address meaning that should an address change, the ADS client does not need to know the new address.

It is possible to manage the acquiring, tracking and releasing of handles yourself, which is advantageous if you plan on reading/writing the value frequently in your program, or wish to speed up the reading/writing by up to three times; as by default when reading/writing by name it makes 3 ADS calls (acquire, read/write, release), where as if you track the handles manually it only makes a single ADS call.

Using the Connection class:

>>> var_handle = plc.get_handle('global.bool_value')
>>> plc.write_by_name('', True, pyads.PLCTYPE_BOOL, handle=var_handle)
>>> plc.read_by_name('', pyads.PLCTYPE_BOOL, handle=var_handle)
True
>>> plc.release_handle(var_handle)

Be aware to release handles before closing the port to the PLC. Leaving handles open reduces the available bandwidth in the ADS router.

Read and write by address

Read and write UDINT variables by address.

>>> import pyads
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
>>> plc.open()
>>> # write 65536 to memory byte MDW0
>>> plc.write(INDEXGROUP_MEMORYBYTE, 0, 65536, pyads.PLCTYPE_UDINT)
>>> # write memory byte MDW0
>>> plc.read(INDEXGROUP_MEMORYBYTE, 0, pyads.PLCTYPE_UDINT)
65536
>>> plc.close()

Toggle bitsize variables by address.

>>> # read memory bit MX100.0
>>> data = plc.read(INDEXGROUP_MEMORYBIT, 100*8 + 0, pyads.PLCTYPE_BOOL)
>>> # write inverted value to memory bit MX100.0
>>> plc.write(INDEXGROUP_MEMORYBIT, 100*8 + 0, not data)

Read and write multiple variables with one command

Reading and writing of multiple values can be performed in a single transaction. After the first operation, the symbol info is cached for future use.

>>> import pyads
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
>>> var_list = ['MAIN.b_Execute', 'MAIN.str_TestString', 'MAIN.r32_TestReal']
>>> plc.read_list_by_name(var_list)
{'MAIN.b_Execute': True, 'MAIN.str_TestString': 'Hello World', 'MAIN.r32_TestReal': 123.45}
>>> write_dict = {'MAIN.b_Execute': False, 'MAIN.str_TestString': 'Goodbye World', 'MAIN.r32_TestReal': 54.321}
>>> plc.write_list_by_name(write_dict)
{'MAIN.b_Execute': 'no error', 'MAIN.str_TestString': 'no error', 'MAIN.r32_TestReal': 'no error'}

Device Notifications

ADS supports device notifications, meaning you can pass a callback that gets executed if a certain variable changes its state. However as the callback gets called directly from the ADS DLL you need to extract the information you need from the ctypes variables which are passed as arguments to the callback function. A sample for adding a notification for an integer variable can be seen here:

>>> import pyads
>>> from ctypes import sizeof
>>>
>>>
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
>>> plc.open()
>>> tags = {"GVL.integer_value": pyads.PLCTYPE_INT}
>>>
>>> # define the callback which extracts the value of the variable
>>> def mycallback(notification, data):
>>>     data_type = tags[data]
>>>     handle, timestamp, value = plc.parse_notification(notification, data_type)
>>>     print(value)
>>>
>>> attr = pyads.NotificationAttrib(sizeof(pyads.PLCTYPE_INT))
>>>
>>> # add_device_notification returns a tuple of notification_handle and
>>> # user_handle which we just store in handles
>>> handles = plc.add_device_notification('GVL.integer_value', attr, mycallback)
>>>
>>> # To remove the device notification use the del_device_notification function.
>>> plc.del_device_notification(handles)
>>> plc.close()

This examples uses the default values for NotificationAttrib. The default behaviour is that you get notified when the value of the variable changes on the server. If you want to change this behaviour you can set the NotificationAttrib.trans_mode attribute to one of the following values:

ADSTRANS_SERVERONCHA (default)
a notification will be sent everytime the value of the specified variable changes
ADSTRANS_SERVERCYCLE
a notification will be sent on a cyclic base, the interval is specified by the cycle_time property
ADSTRANS_NOTRANS
no notifications will be sent

For more information about the NotificationAttrib settings have a look at Beckhoffs specification of the AdsNotificationAttrib struct.

Device Notification callback decorator

To make the handling of notifications more pythonic a notification decorator has been introduced in version 2.2.4. This decorator takes care of converting the ctype values transferred via ADS to python datatypes.

>>> import pyads
>>> plc = pyads.Connection('127.0.0.1.1.1', 48898)
>>> plc.open()
>>>
>>> @plc.notification(pyads.PLCTYPE_INT)
>>> def callback(handle, name, timestamp, value):
>>>     print(
>>>         '{1}: received new notitifiction for variable "{0}", value: {2}'
>>>         .format(name, timestamp, value)
>>>     )
>>>
>>> plc.add_device_notification('GVL.intvar', pyads.NotificationAttrib(2),
                                callback)
>>> # Write to the variable to trigger a notification
>>> plc.write_by_name('GVL.intvar', 123, pyads.PLCTYPE_INT)

2017-10-01 10:41:23.640000: received new notitifiction for variable "GVL.intvar", value: abc

Structures can be read in a this way by requesting bytes directly from the PLC. Usage is similar to reading structures by name where you must first declare a tuple defining the PLC structure.

>>> structure_def = (
...     ('rVar', pyads.PLCTYPE_LREAL, 1),
...     ('rVar2', pyads.PLCTYPE_REAL, 1),
...     ('iVar', pyads.PLCTYPE_INT, 1),
...     ('iVar2', pyads.PLCTYPE_DINT, 3),
...     ('sVar', pyads.PLCTYPE_STRING, 1))
>>>
>>> size_of_struct = pyads.size_of_structure(structure_def)
>>>
>>> @plc.notification(size_of_struct)
>>> def callback(handle, name, timestamp, value):
...     values = pyads.dict_from_bytes(value, structure_def)
...     print(values)
>>>
>>> attr = pyads.NotificationAttrib(ctypes.sizeof(size_of_struct))
>>> plc.add_device_notification('global.sample_structure', attr, callback)

OrderedDict([('rVar', 11.1), ('rVar2', 22.2), ('iVar', 3), ('iVar2', [4, 44, 444]), ('sVar', 'abc')])

The notification callback works for all basic plc datatypes but not for arrays. Since version 3.0.5 the ctypes.Structure datatype is supported. Find an example below:

>>> class TowerEvent(Structure):
>>>     _fields_ = [
>>>         ("Category", c_char * 21),
>>>         ("Name", c_char * 81),
>>>         ("Message", c_char * 81)
>>>     ]
>>>
>>> @plc.notification(TowerEvent)
>>> def callback(handle, name, timestamp, value):
>>>     print(f'Received new event notification for {name}.Message = {value.Message}')

Symbols

Symbol creation

Reading from or writing to an ADS variable (= an ADS symbol) can be done even more pythonic through an AdsSymbol instance:

>>> import pyads
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
>>> plc.open()
>>> symbol = plc.get_symbol('global.bool_value')

The address and type of the symbol will be automatically determined using a READ_WRITE request to the ADS server, based on the variable name. This lookup is skipped when all the information has already been provided:

>>> import pyads
>>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1)
>>> plc.open()
# Remaining info will be looked up:
>>> symbol = plc.get_symbol('global.bool_value')
# Alternatively, specify all information and no lookup will be done:
>>> symbol = plc.get_symbol('global.bool_value', index_group=123,
                            index_offset=12345, symbol_type=pyads.PLCTYPE_BOOL)

Here the indices are same as used in Connection.read() and Connection.write(). The symbol type can also be the same as with the read and write method, e.g. pyads.PLCTYPE_INT or pyads.PLCTYPE_BOOL. Alternatively, the class will also accept a string of the variable type in PLC-style, e.g. ‘LREAL’, ‘INT’, ‘UDINT’, etc.

Symbols also work with structures and arrays of structures. Use the parameter structure_def to define the structure and array_size to define the size of the array.

 >>> structure_def = (
         ("i", pyads.PLCTYPE_INT, 1),
         ("s", pyads.PLCTYPE_STRING, 1)
     )
 >>> symbol = plc.get_symbol("MyStructure", structure_def=structure_def, array_size=2)
 >>> symbol.write([{"i": 1, " "s": "foo"}, {"i": 2, "s": "bar"}])
 >>> symbol.read()
[{"i": 1, " "s": "foo"}, {"i": 2, "s": "bar"}]

Read and write operations

Reading from and writing to symbols is straightforward:

>>> symbol.read()
True
>>> symbol.write(False)
>>> symbol.read()
False
>>> plc.close()

The symbol objects have the value property, which is the buffered symbol value:

>>> if symbol.read() > 0.5:
>>>     print(symbol.value)

The example above will perform a single READ request. value is updated on every read and write of the symbol. If None is passed to AdsSymbol.write() (the default parameter), the buffer will be written to the target:

>>> symbol.write(3.14)

>>> # Is identical to:
>>> symbol.value = 3.14
>>> symbol.write()

The symbol can be set to auto-update the AdsSymbol.value property through a device notification. See the subsection below.

Device notifications

Notifications (function callbacks) can be attached directly to a symbol:

>>> symbol.add_device_notification(my_func)

The symbol will track the handles of the notifications attached to it and free them up when the object runs out of scope.

You can delete specific notifications or clear all of them:

>>> handles = symbol.add_device_notification(my_func)
>>> symbol.del_device_notification(handles)

>>> # Or clear all:
>>> symbol.clear_device_notifications()

AdsSymbol.add_device_notification() will automatically create a notification attribute object with the right variable length. You can also specify an optional notification attribute and/or user handle:

>>> attr = NotificationAttrib(length=sizeof(pyads.PLCTYPE_BOOL), max_delay=1.0, cycle_time=1.0)
>>> user_handle = 123
>>> symbol.add_device_notification(my_func, attr=attr, user_handle=user_handle)

Auto-update

A built-in notification is available to automatically update the symbol buffer based on the remote value. This is disabled by default, enable it with:

>>> symbol.auto_update = True

This will create a new notification callback to update AdsSymbol.value. This can be efficient if the remote variable changes less frequently then your code runs. The number of notification callbacks will then be less than what the number of read operations would have been.

It can be disabled again with:

>>> symbol.auto_update = False

Using auto_update will also write the value immediately to the plc when AdsSymbol.value` is changed.

Warning

Take care that AdsSymbol.clear_device_notifications() will also remove the auto-update notification. Like all symbol notifications, the auto-update will be cleared automatically in the object destructor.

Get all symbols

In order to get a list of the device’s declared variables, use the get_all_symbols method.

>>> symbols = plc.get_all_symbols()
>>> print('\n'.join("%s: %s" % item for item in vars(symbols[0]).items()))
index_group: 16448
index_offset: 384800
name: Constants.bFPUSupport
symtype: BOOL
comment: Does the target support multiple cores?

Testserver

The testserver was created initially for internal testing. However, you can also use the testserver to test your application. Use it when no real ADS server is available, for example during continuous integration or when TwinCAT is not installed.

You can run a basic testserver with the command:

$ python -m pyads.testserver --handler basic

The handler type defaults to ‘advanced’.

This will create a new device on 127.0.0.1 port 48898. In the next step the route to the testserver needs to be added from another python console.

>>> import pyads
>>> pyads.add_route("127.0.0.1.1.1", "127.0.0.1")

Warning

The testserver functionality was originally intended only for internal testing. The documentation and any support are not guaranteed.

Handlers

The server is a socket.socket listener, that listens to ADS-like connections and sends responses to requests. AdsTestServer itself does not manage the requests and responses. Those are managed by handler classes. Currently there are two handlers available:

BasicHandler always returns the same static responses. No data can be saved, any returned values are always 0.

AdvancedHandler keeps a list of variables and allows for reading/writing variables. Variables need to be created upfront via add_variable().

Your requirements determine which handler is most suitable. You can also create your own handler by extending the AbstractHandler class. Typically, the basic handler will require the least amount of work.

A complete overview of the capabilities of the handlers is below. If a feature is mocked, it will do nothing but no error will be thrown when it is executed. If a feature is not implemented, an error will be thrown when an attempt is made to use the feature.

Handler implementations
Feature (Methods from `Connection`)	`BasicHandler`	`AdvancedHandler`
read_state	Mocked	Mocked
write_control	Mocked	Mocked
read_device_info	Mocked	Mocked
read	Mocked	Implemented
write	Mocked	Implemented
read_by_name	Mocked	Implemented
read_by_name (with handle)	Mocked	Implemented
write_by_name	Mocked	Implemented
write_by_name (with handle)	Mocked	Implemented
get_symbol	Mocked (no info will be found automatically)	Implemented
get_all_symbols	Mocked (list will always be empty)	Implemented
get_handle	Mocked	Implemented
release_handle	Mocked	Mocked
read_list_by_name	Mocked	Implemented
write_list_by_name	Mocked	Implemented
read_structure_by_name	Mocked	Not implemented
write_structure_by_name	Mocked	Not implemented
add_device_notification	Mocked	Implemented
del_device_notification	Mocked	Implemented
Device notifications	Not implemented (callbacks will never fire)	Implemented

Basic Handler

The BasicHandler just responds with 0x00 wherever possible. Trying to read any byte or integer will always always net 0. Trying to read an LREAL for example will give 2.09e-308, as that is the interpretation of all bits at 0.

Actions like writing to a variable or adding a notification will always be successful, but they won’t have any effect.

Advanced Handler

The AdvancedHandler keeps track of variables in an internal list. You can read from and write to those variables like you would with a real server, using either the indices, name or variable handle. Any notifications will be issued as expected too. The handler keeps a list of variables with the type PLCVariable. In order to address a variable you need to explicitly create it first:

# Server code

handler = AdvancedHandler()

test_var = PLCVariable(
    "Main.my_var", bytes(8), ads_type=constants.ADST_REAL64, symbol_type="LREAL"
)
handler.add_variable(test_var)

# Client code

with plc:
    sym = plc.get_symbol("Main.my_var")  # Already exists remotely
    print(sym)
    print(sym.read())

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

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

Welcome to pyads’s documentation!

Installation

Installation on Linux

Installation on Windows

Testing your installation

Quickstart

Documentation

Routing

Creating routes on Windows

Creating routes on Linux

Adding routes to a target

Connections

Connect to a remote device

Read and write by name

Values

Arrays

Structures of the same datatype

Structures with multiple datatypes

Read and write by handle

Read and write by address

Read and write multiple variables with one command

Device Notifications

Device Notification callback decorator

Symbols

Symbol creation

Read and write operations

Device notifications

Auto-update

Get all symbols

Testserver

Handlers

Basic Handler

Advanced Handler

pyads package

pyads.ads module

pyads.constants module

pyads.connection module

pyads.errorcodes module

pyads.pyads_ex module

pyads.structs module

pyads.symbol module

pyads.testserver module

pyads.testserver.testserver

pyads.testserver.handler

pyads.testserver.basic_handler

pyads.testserver.advanced_handler

pyads.utils module

Module contents

Indices and tables