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 :py:class:`~sila2.server.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. .. code-block:: python # 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 :py:class:`~typing.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=...)``. .. code-block:: python # 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: .. code-block:: python # 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 :py:func:`FeatureImplementationBase.run_periocially() `. .. code-block:: python 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 :py:class:`~sila2.server.ObservableCommandInstance` object. It can be used to send updates to the client while the command is running (:py:attr:`~sila2.server.ObservableCommandInstance.progress`, :py:attr:`~sila2.server.ObservableCommandInstance.estimated_remaining_time`, :py:attr:`~sila2.server.ObservableCommandInstance.lifetime_of_execution`, and :py:func:`~sila2.server.ObservableCommandInstanceWithIntermediateResponses.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. .. code-block:: python 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, :py:class:`~sila2.framework.DefinedExecutionError` subclasses are generated. You can import them from the generated feature submodule and raise them like normal Python exceptions. .. code-block:: python 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 :py:class:`~sila2.framework.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 :py:class:`~sila2.server.MetadataDict`. When SiLA clients send metadata along with a request for property access or command execution, this metadata will be included in that :py:class:`dict`-like object. The keys are :py:class:`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 :py:class:`~sila2.server.MetadataInterceptor` and adding it to the ``Server`` class. .. code-block:: python # 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 .. code-block:: python # 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 :py:func:`~sila2.server.FeatureImplementationBase.start` and :py:func:`~sila2.server.FeatureImplementationBase.stop` methods inherited from :py:class:`~sila2.server.FeatureImplementationBase`. These methods are called by the server, do not call them yourself. Make sure to call the super methods: .. code-block:: python def start(self) -> None: print("Feature implementation started") super().start() def stop(self) -> None: print("Feature implementation stopped") super().stop()