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 False TLS/SSL certificate verification is switched off. Default is True.

async close() → None: Close the connection.

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 False TLS/SSL certificate verification is switched off. Default is True.

async read(element: str) → Any

Read the current value.

Returns:: the current value
Return type:: Any

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:

Subscription

async write(element: str, value: Any) → Any

Write a value.

Parameters:: value – the value to be written

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.

async read() → Any

Read the current value.

Returns:: the current value
Return type:: Any

async write(value: Any) → Any

Write a value.

Parameters:: value – the value to be written

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)

async receive() → Event: Receive the next event in the subscription.

async start() → None: Start the subscription.

async stop() → None: Stop the subscription.

Event

class htf.xentara.Event(timestamp: float, name: str, uuid: UUID, value: Any)

Yielded in Subscription when used as an async iterator.

name: str: the name of the element

timestamp: float: the timestamp as a float

uuid: UUID: the uuid of the element

value: Any: the current value of the element