Implementing SiLA features¶

For each feature that is added to a SiLA Server/Client package, a file [featureidentifier]_impl.py is generated in the feature_implementations directory. It contains a class [FeatureIdentifier]Impl, which extends the generated base class [FeatureIdentifier]Base. This base class extends the class FeatureImplementationBase.

All code examples follow the features implemented in the example server.

Unobservable property¶

For each unobservable property, a method get_[PropertyIdentifier](self, *, metadata) is generated. It is called when a SiLA Client requests the current property value.

# GreetingProvider.StartYear
def get_StartYear(self, *, metadata) -> int:
    return 2022

Unobservable command¶

For each unobservable command, a method [CommandIdentifier](self, Param1, Param2, ..., *, metadata) is generated. It is called when a SiLA Client requests the execution of this command. Each parameter from the feature definition is translated into one method argument (e.g. Param1). You have to return a [CommandIdentifier]_Responses object, which is a NamedTuple instance with one field per command response specified in the feature definition. Example: The command SayHello has one response “Greeting”. To implement this method, use return SayHello_Responses(Greeting=...).

# GreetingProvider.SayHello
def SayHello(self, Name: str, *, metadata) -> SayHello_Responses:
    return SayHello_Responses(Greeting="Hello " + Name)

Observable property¶

When a SiLA Client requests to subscribe to an observable property, the method [PropertyIdentifier]_on_subscription(self, *, metadata) is called. It is defined in the Base class. If you want to specify what happens on subscriptions, override this method, and don’t forget to call the super method:

# TimerProvider.CurrentTime: print a message when a client subscribes
def CurrentTime_on_subscription(self, *, metadata) -> None:
    print("A client subscribed to MyProperty!")
    super().MyProperty_on_subscription(metadata=metadata)

To update the property value, use the method update_MyProperty(new_value), which is also defined in the Base class. When this method is called, the new value is sent to all currently subscribed clients, and it is stored and sent immediately when a clients subscribes to the property.

You can call this method from anywhere, e.g. in the __init__ method of the class to set an initial value, and when other methods are called that change the value of the property.

If you regularly want to update the value, you can use the method FeatureImplementationBase.run_periocially().

from datetime import datetime, time, timezone

# TimerProvider.CurrentTime: update observable property every second
class TimerProviderImpl(TimerProviderBase):
    def __init__(self, parent_server: Server) -> None:
        super().__init__(parent_server=parent_server)

        def update_current_time() -> None:
            # get current time
            current_time: time = datetime.now(tz=timezone.utc).timetz()

            # update observable property
            self.update_CurrentTime(current_time)

        self.run_periodically(update_current_time, delay_seconds=1)

Observable command¶

Observable commands are implemented similarly to unobservable commands. A method [CommandIdentifier](self, Param1, Param2, ..., *, metadata, instance) is generated, which is called when a SiLA Client requests to execute the command. Like command responses, intermediate responses are also represented by :py:class:~typing.NamedTuple` classes.

There are two additional things to keep in mind:

Command instance¶

The implementation method receives the additional parameter instance, which is an ObservableCommandInstance object. It can be used to send updates to the client while the command is running (progress, estimated_remaining_time, lifetime_of_execution, and send_intermediate_response()).

Lifetime of execution¶

Instances of observable commands have a “lifetime of execution”. After its lifetime, client cannot subscribe to the status or intermediate responses of the instance. The lifetime can be extended, but not reduced.

Warning

When the lifetime is None, it is infinite. This means that the command instance is stored in the server memory until the server is shut down.

class TimerProviderImpl(TimerProviderBase):
    def __init__(self, parent_server: Server) -> None:
        ...
        self.Countdown_default_lifetime_of_execution = datetime.timedelta(seconds=2)

    def Countdown(
        self, N: int, Message: str, *, metadata, instance: ObservableCommandInstanceWithIntermediateResponses[Countdown_IntermediateResponses]
    ) -> Countdown_Responses:
        # change status from `waiting` to `running`
        instance.begin_execution()

        instance.progress = 0
        instance.estimated_remaining_time = datetime.timedelta(seconds=N)
        instance.lifetime_of_exection = datetime.timedelta(2 + N)

        for i in range(N, 0, -1):
            instance.send_intermediate_response(Countdown_IntermediateResponses(i))  # send intermediate responses
            instance.progress = (N - i) / N * 100
            instance.estimated_remaining_time = datetime.timedelta(seconds=i)

            time.sleep(1)

        return Countdown_Responses(Message, datetime.datetime.now(tz=datetime.timezone.utc))

Error handling¶

Defined Execution Errors¶

If the feature definition contains defined execution errors, DefinedExecutionError subclasses are generated. You can import them from the generated feature submodule and raise them like normal Python exceptions.

from ..generated.timerprovider import CountdownTooLong

class TimerProviderImpl(TimerProviderBase):
    def Countdown(self, N: int, ...):
        if N > 9000:
            raise CountdownTooLong

Undefined Execution Errors¶

If your code raises an uncaught exception, the SDK will wrap it in an UndefinedExecutionError and sent that to the client.

Validation Errors¶

These are raised by the SDK automatically if a client sends parameters that violate constraints specified in the feature definition. In such cases, your feature implementation code will not be called.

Metadata¶

Receiving metadata in command/property implementations¶

The property access/subscription and also the command methods have a parameter metadata, which is a MetadataDict. When SiLA clients send metadata along with a request for property access or command execution, this metadata will be included in that dict-like object. The keys are sila2.framework.Metadata objects and the values are the metadata values sent by the client.

Metadata Interceptors¶

Usually, metadata affects multiple commands and properties and should be handled in the same way. This can be achieved by implementing a MetadataInterceptor and adding it to the Server class.

# new file feature_implementations/delay_interceptor.py
from typing import Any, Dict

from sila2.framework import FullyQualifiedIdentifier
from sila2.server import MetadataDict, MetadataInterceptor

from ..generated.delayprovider import DelayProviderFeature, DelayTooLong

class DelayInterceptor(MetadataInterceptor):
    def __init__(self):
        # this interceptor affects the Delay metadata in the DelayProvider feature
        super()__init__(affected_metadata=[DelayProviderFeature["Delay"]])

    def intercept(self, parameters, metadata: MetadataDict, target_call: FullyQualifiedIdentifier) -> None:
        delay: int = metadata[DelayProviderFeature["Delay"]]

        if delay > 10_000:
            raise DelayTooLong("Maximum delay is 10 seconds")  # defined execution error

        time.sleep(delay/1000)  # delay is given in milliseconds

# in server.py
from .feature_implementations.delay_importceptor import DelayInterceptor

class Server(SilaServer):
    def __init__(...):
        ...

        # add interceptor to server
        self.add_metadata_interceptor(DelayInterceptor())

`start()` and `stop()`: Setup and Shutdown¶

If you want to execute code on server startup or shutdown, you can override the start() and stop() methods inherited from FeatureImplementationBase. These methods are called by the server, do not call them yourself.

Make sure to call the super methods:

def start(self) -> None:
    print("Feature implementation started")
    super().start()

def stop(self) -> None:
    print("Feature implementation stopped")
    super().stop()

Implementing SiLA features¶

Unobservable property¶

Unobservable command¶

Observable property¶

Observable command¶

Command instance¶

Lifetime of execution¶

Error handling¶

Defined Execution Errors¶

Undefined Execution Errors¶

Validation Errors¶

Metadata¶

Receiving metadata in command/property implementations¶

Metadata Interceptors¶

start() and stop(): Setup and Shutdown¶

`start()` and `stop()`: Setup and Shutdown¶