Showing results for 
Search instead for 
Did you mean: 

Python Runtime App: is subscribe() to datalayer nodes possible?

Python Runtime App: is subscribe() to datalayer nodes possible?

New Contributor

Is it possible to use subscription() to datalayer nodes inside a *.py file?

Or could we only use and we have to poll it?



Hi @LunatiX ,

Absolutely! See the python datalayer client subscription sample in the SDK.

Please let me know if you have any questions on the example.

New Contributor

When I build a own python app, then it's clear for me. But I would like to use the Python Runtime App with a simple *.py file and I only see this commands here:

  • datalayer.browse
  • datalayer.write
  • datalayer.create
  • datalayer.remove
  • datalayer.read_json
  • datalayer.write_json
  • datalayer.create_json

See: Script Parser/Interpreter (Python) - Bosch Rexroth Produktinformationsportal

See my short example too, which is working well with and datalayer.write():


But the question is, are there any commands to subscribe?

@LunatiX ,

The examples have multiple files, but they can be run outside of the snap context. See this article on running the samples directly in the IDE app.

The library you are using does not have a subscribe function.

Occasional Contributor

The code below, which creates a subscription and dumps the result into the diagnostics log, is similar to the source code of the library in question.  You can run this directly with the Python Runtime app.  

import requests
from urllib3.exceptions import InsecureRequestWarning


session = requests.Session()
session.verify = False

response ="https://localhost/identity-manager/api/v2/auth/token", json={"name":"boschrexroth","password":"boschrexroth"}).json()

token = response['access_token']
data = {"Authorization": f"Bearer {token}"}

with session.get("https://localhost/automation/api/v2/events?nodes=framework/metrics/system/cpu-utilisation-percent&publishIntervalMs=5000", headers=data, stream=True) as stream:
    for line in stream.iter_lines():
        tmp = line.decode("utf-8").split(":",1)
        if len(tmp) > 1 and tmp[0] == 'data':
            print(tmp[1], flush=True)



I tried a simillar aproche like @LunatiX without success:

Starting point is the documentation here


I tried to use the subscription methods from the ctrlxdatalayer library:
create_properties to generate the Rulset
and then the subscription_async class for the subscription

However, I already run into a problem with the following simple code:


from ctrlxdatalayer import subscription

print("Make subscription to one node")
itemProperty = subscription.create_properties("opcuaclient/SPS/i=85/ns=2;s=Logic/ns=2;s=.I40_PartIndetifier", 100, 60000, 10000)
print("Property done")



  • Running the code in the IDE as single file (Debug: Python: current file): All fine, no error.
  • Running the code in the Python script engine (Debug: Launch Python in Script Egnine):
    OSError(' cannot open shared object file: No such file or directory')

Is this the correct way to establish a subscription to multiple data layer items?
What am I missing?



Occasional Contributor

@Reto :

From the documentation for the Python Runtime Environment:


The library providing datalayer access that is integrated into the Python Runtime App - the one that @LunatiX refers to in his original post - is built on REST calls using the requests module.  Because ctrlxdatalayer requires, as far as I know there is no way to use it in this context.

To create multiple subscriptions in a single REST call, see @CodeShepherd's post here.


Thanks for the clarification. I was aware of the binary limitation, but thought it only applied to external, user-uploaded libraries, not those supplied by Rexroth.