Unitelabs SiLA Python Library

A Python library for creating SiLA 2 clients and servers. This flexible and unopinionated library gives you everything needed to create a SiLA 2 1.1 compliant Python application. It adheres to the SiLA 2 specification and is used by the UniteLabs CDK to enable rapid development of cloud-native SiLA Servers with a code-first approach.

Getting Started

Prerequisites

Ensure you have Python 3.10+ installed. You can install Python from python.org.

Quickstart

To get started quickly with your first connector, we recommend to use our UniteLabs CDK. Use Cookiecutter to create your project base on our Connector Factory starter template:

cookiecutter git@gitlab.com:unitelabs/cdk/connector-factory.git

Installation

Install the latest version of the library into your Python project:

pip install unitelabs-sila

Usage

Client

Connect to a server and call its features by their identifiers:

import asyncio

from sila.client import Client, ClientConfig

async def main():
    async with Client(ClientConfig(hostname="localhost", port=50052, tls=False)) as client:
        responses = await client.GreetingProvider.SayHello(Name="World")
        print(responses.Greeting)

asyncio.run(main())

Once the client is open, every feature the server implements is an attribute named after its identifier, like client.GreetingProvider. Its properties, commands, metadata, and defined execution errors are attributes of that feature.

Properties

year = await client.GreetingProvider.StartYear.read()

async for value in await client.ObservablePropertyTest.Alternating.subscribe():
    print(value)

Commands

Pass parameters as keyword arguments. An unobservable command returns its responses:

responses = await client.GreetingProvider.SayHello(Name="World")
responses.Greeting     # "Hello World"

An observable command returns its execution. Intermediate and final responses work the same way:

execution = await client.ObservableCommandTest.Count(N=5, Delay=1)

async for response in await execution.subscribe_intermediate_responses():
    print(response.CurrentIteration)

responses = await execution.get_responses()

Metadata, errors, and timeouts

metadata and timeout are keyword arguments on every call. Metadata and defined execution errors come from their feature:

lock = await client.LockController.LockIdentifier.from_native(client, "my-lock")
await client.GreetingProvider.SayHello(Name="World", metadata=[lock], timeout=5000)

try:
    await client.LockController.LockServer(LockIdentifier="my-lock", Timeout=60)
except client.LockController.ServerAlreadyLocked:
    print("Someone else holds the lock.")

Features with the same identifier

A server can implement several features with the same identifier, from different originators or in different major versions. client.LockController then raises an AmbiguousIdentifierError that lists them. Pick one with client.feature():

lock = client.feature("LockController", originator="org.silastandard")
lock = client.feature("LockController", version=2)
lock = client.feature("org.silastandard/core/LockController/v2")

await lock.LockServer(LockIdentifier="my-lock", Timeout=60)

Every other feature keeps its shortcut. A property and a command that share an identifier within one feature raise the same error. Get each by its fully qualified identifier:

flow_rate = client.get_property("com.example/devices/Pump/v1/Property/FlowRate")
set_flow_rate = client.get_command("com.example/devices/Pump/v1/Command/FlowRate")

Errors and timeouts

Every error the client raises while talking to a server is a SiLAError. Catch its subclasses to tell them apart:

• SiLATimeoutError: a timeout elapsed. It's a SiLAConnectionError, so catch it first.
• SiLAConnectionError: the server is unavailable or the connection failed. status_code holds the gRPC status code, if the error comes from a call.
• ValidationError: a command parameter is missing or invalid. parameter names it.
• DefinedExecutionError: an error the feature declares in its definition.
• UndefinedExecutionError: any other failure on the server, including a defined execution error the client doesn't know.
• FrameworkError and BinaryTransferError: the SiLA protocol was used incorrectly, or a large binary failed to transfer. Their subclasses, like InvalidMetadata, tell which. CommandExecutionNotAccepted means the command was not performed: the server refused it, or the client never sent it. In the second case, its __cause__ is the connection error.

Timeouts are in milliseconds. A request without a timeout uses the client's default_timeout, and waits indefinitely when neither is set. A timeout on a subscription only bounds setting it up, not how long it runs. ObservableProperty.read() is the exception: its timeout also bounds the wait for the first value.

A command whose timeout runs out before it was sent raises CommandExecutionNotAccepted instead. So does a command on a connection that is already closed, or whose binary parameter failed to upload. The command was not performed, so running it again is safe. Once the client handed a command to gRPC, a SiLAConnectionError or SiLATimeoutError can't tell whether the server got it, so the server may have run it: check its state before running it again.

from sila.client import SiLAConnectionError, SiLATimeoutError

try:
    await client.open(timeout=5000)
except SiLATimeoutError:
    print("The server didn't answer within 5 s.")
except SiLAConnectionError as error:
    print(f"The connection failed while loading the features: {error}")

Server

To start using the SiLA Python library as a server:

1. Import and configure your SiLA server instance:

   import asyncio
   
   from sila.server import Server
   from your_project.features import your_feature
   
   async def main():
       server = Server({"port": 50000})
       server.add_feature(your_feature)
       await server.start()
   
   asyncio.run(main())
   

2. To implement a custom SiLA Feature, create a feature definition following the SiLA2 specification:

   from sila.server import Feature, UnobservableCommand
   
   your_feature = Feature(...)
   your_method = UnobservableCommand(...)
   your_method.add_to_feature(your_feature)
   

3. Run your server:

   python your_script.py
   

Important: Without implementing the required SiLA Service Feature, your SiLA Server will not be fully compliant with the standard. For easier compliance, consider using the UniteLabs CDK, which handles this automatically.

Contribute

Submit and share your work at https://hub.unitelabs.io.

We encourage you to submit feature requests and bug reports through the GitLab issue system. Please include a clear description of the issue or feature you are proposing. If you have further questions, issues, or suggestions for improvement, don't hesitate to reach out to us at developers@unitelabs.io.

Join the conversation! Stay up to date with the latest developments by joining the Python channel in the SiLA Slack.

License

Distributed under the MIT License. See MIT license for more information.