Xentara โ Real-Time Automation Platform๏
Introduction๏
Xentara is a modular, secure and powerful Real-Time Automation Platform. You can find more information at the Xentara website.
htf includes a websocket-client to interact with the Xentara runtime.
Thus, users can for example deploy tests on their machines and execute tests remotely.
The Xentara-client is async-only.
Open a Connection๏
To open a connection, instantiate XentaraClient and await it.
client = await XentaraClient(
url="wss://xentara-host/api/ws",
root_element="Hardware",
)
Note
Donโt forget the await statement!
Alternatively use create.
client = await XentaraClient.create(
url="wss://xentara-host/api/ws",
root_element="Hardware",
)
The root element (root_element) must match the configuration Xentara was started with.
The contents are browsed and fetched automatically on instantiation.
Authentication๏
The Xentara-client supports basic authentication.
The credentials are simply added to the url, i.e.:
client = await XentaraClient(
url="wss://<user>:<password>@xentara-host/api/ws",
)
where <username> is the username and <password> is the password to be used.
TLS/SSL๏
Use wss:// in the url for a TLS/SSL secured connection and ws:// for an unsafe and unencrypted connection.
Self-signed certificates๏
To disable the TLS/SSL certificate verification, just add verify_ssl=False.
client = await XentaraClient(
url="wss://<user>:<password>@xentara-host/api/ws",
root_element="Hardware",
verify_ssl=False,
)
Close a Connection๏
To close an open connection, simply run:
await client.close()
Reading values๏
To read a value, call read:
value = await client.read("<element>")
<element> is a descriptor of a Xentara element, for example "Hardware.EtherCAT Bus.EL2624.Output 1".
In addition, the Xentara-client can be used like a dictionary to read values.
value = await client["<element>"].read()
Writing values๏
To write a value, call write:
await client.write("<element>", value)
<element> is a descriptor of a Xentara element, for example "Hardware.EtherCAT Bus.EL2624.Output 1".
The Xentara-client can be used like a dictionary to write values.
await client["<element>"].write(value)
Subscriptions๏
To subscribe to changes, use the subscribe function that returns a
Subscription.
The htf.xentara.Subscription returns an async iterable that can be used in different ways.
Each iteration yields an Event.
The easiest way to use a subscription is:
async with client.subscribe("<element>") as subscription:
for i in range(5):
event = await subscription.receive()
value = event.value
print(f"{i}: {event=}")
aiter can also be used explicitly:
async with client.subscribe("<element>") as subscription:
iterator = aiter(subscription)
for i in range(5):
event = await iterator.__anext__()
value = event.value
print(f"{i}: {event=}")
For an infinite number of received values, async for can be used.
subscription = client.subscribe("<element>")
async with subscription as subscription:
async for event in subscription:
value = event.value
print(f"{i}: {event=}")
String Representation๏
The Xentara client can be printed to get information about it.
print(client)
Use the Client like a Dictionary๏
To get the available elements, the XentaraClient can be used like a dictionary.
keys = client.keys()
values = client.values()
items = client.items()
length = len(client)
XentaraClient๏
- class htf.xentara.XentaraClient(url: str, root_element: str = 'Hardware', verify_ssl: bool = True)๏
The Xentara websocket client.
- Parameters:
url โ The url to connect to the Xentara server.
root_element โ The root element to browse for elements. This is set in the Xentara configuration.
verify_ssl โ if set to
FalseTLS/SSL certificate verification is switched off. Default isTrue.
- async copy() XentaraClient๏
Return a copy of the instance with a new websocket connection.
- async classmethod create(url: str, root_element: str = 'Hardware', verify_ssl: bool = True) XentaraClient๏
Create the Xentara client. Is called when the class instance is awaited, too.
- Parameters:
url โ The url to connect to the Xentara server.
root_element โ The root element to browse for elements. This is set in the Xentara configuration.
verify_ssl โ if set to
FalseTLS/SSL certificate verification is switched off. Default isTrue.
- subscribe(element: str, maximum_update_interval: int = 500, minimum_update_interval: int = 1000) Subscription๏
Create a subscription.
- Parameters:
element โ The element to be subscribed.
maximum_update_interval โ the maximum update interval in milliseconds (default: 500)
minimum_update_interval โ the minimum update interval in milliseconds (default: 1000)
- Returns:
the instance of the subscription
- Return type:
Element๏
- class htf.xentara.Element(client: XentaraClient, name: str)๏
An element to read and write values. Used when :class:~`htf.xentara.XentaraClient` is used like a dictionary.
Subscription๏
- class htf.xentara.Subscription(client: XentaraClient, element: str, maximum_update_interval: int = 500, minimum_update_interval: int = 1000)๏
Subscriptions for Xentara elements.
Users will never call this method directly.
- Parameters:
client โ the current Xentara clientโs instance (passed automatically)
element โ the name of the element the subscription is for
maximum_update_interval โ the maximum update interval in milliseconds (default: 500)
minimum_update_interval โ the minimum update interval in milliseconds (default: 1000)