Data Endpoints

Data endpoints represent the subset of SiLA endpoints which return data to the user. Observable Properties and Unobservable Properties as well as Unobservable Commands may all be used to create Data Endpoints depending on the function being wrapped.

In this context, properties are relatively clear in their data extraction functionality. Unobservable commands, however, could be functions which have input parameters (which are not allowed for SiLA properties) and use those parameters to selectively get information from the device, i.e. extracting a single value from a dictionary based on its key.

The following use case exemplifies the distinction we want to make between data endpoints and controls.

@sila.UnobservableCommand()
@sila.Response(name="Value")
async def get_variable_value(self, variable: str) -> str:
    """
    Read out the variable value for the currently active configuration.
    
    .. parameter: The name of the variable in the active configuration
      to get the value for.
    .. return:: The variable value.
    """

    return self.active_config.get(variable, "")

The above function does not fit into the SiLA definition of a property because it is parameterized, however its functionality more closely matches that of properties conceptually as static information providers.

With this distinction in mind, we will proceed to a real world example and implement some data endpoints.

Implementation Example

Throughout this and the next tutorial we will use the example of a clock to understand the differences between different SiLA methods. We will implement a SiLA Clock Feature and then create data endpoints for each of the three types.

Making a feature class

For our feature class we will start by declaring methods that we will later implement on the feature.

import asyncio
import typing

from unitelabs.cdk import sila
from unitelabs.cdk.sila import datetime


class ClockController(sila.Feature):
    def __init__(self):
        super().__init__(
            originator="io.unitelabs",
            category="clock",
            version="1.0",
            maturity_level="Draft",
        )
        self._tz: typing.Optional[datetime.tzinfo] = None

    @sila.UnobservableProperty()
    async def get_current_timestamp(self) -> datetime.datetime:
        """Get the current time."""
        ...

    @sila.ObservableProperty(display_name="Running Clock")
    async def subscribe_current_timestamp(self) -> sila.Stream[datetime.datetime]:
        """Subscribe to the current time."""
        ...

    @sila.UnobservableCommand()
    @sila.Response(name="Time Difference")
    async def get_time_difference(self, future_time: datetime.time) -> int:
        """
        Get the difference in seconds between given `future_time` and the
        current time.

        .. parameter:: The future time to compare the current time with.
        .. return:: The time difference in seconds.
        """
        ...

For each method, we have so far defined:

• name
• method type (Observable vs. Unobservable, Property vs. Command)
• parameters, if any
• return values, if any
• type hints (type of the parameters & return variables)
• documentation (doc string used to specify functionality in a human-readable way)

UnobservableProperty Implementation

The first method we will implement is the unobservable property get_current_timestamp. Here is the stub we produced above.

@sila.UnobservableProperty()
async def get_current_timestamp(self) -> datetime.datetime:
    """Get the current time."""
    ...

In this case, the structure that has been defined is:

• name: get_current_timestamp
• method type: unobservable property
• parameters: only the required self (properties don't allow input parameters)
• returns: a single return value named "Current Timestamp" (based on the name of the function - "get")
• type hints: the return value is a datetime.datetime.
• documentation: everything within the """ ... """. Additional documentation is inferred from the method name and type hints.

Now to add the logic to make this functional:

@sila.UnobservableProperty()
async def get_current_timestamp(self) -> datetime.datetime:
    """Get the current time."""
    return datetime.datetime.now(self._tz)

We use the CDK's SiLA data types to convert our datetime into a SiLA-compatible data type. Here we are using the class's _tz attribute to set the timezone for our timestamp. By default our clock feature has no timezone, but users will be able to set this with the UnobservableCommand set_timezone which we will define later in the Control tutorial, meaning that our feature holds a particular state which can only be modified through usage of the feature.

ObservableProperty Implementation

The observable property we will implement is a method to subscribe to the clock's state, i.e. the current time. Subscribing means retrieving the state and continually receiving the updated state as soon as it changes.

@sila.ObservableProperty(display_name="Running Clock")
async def subscribe_current_timestamp(self) -> sila.Stream[datetime.datetime]:
    """Subscribe to the current time."""
    ...

One important thing to note here is the use of displayname with the sila.ObservableProperty decorator. Without explicitly setting the display_name here, this function would have the same identifier as our UnobservableProperty get_current_timestamp. Remember that the "get" of UnobservableProperties and the "subscribe_" of ObservableProperties are by default stripped from names when creating unique identifiers for methods in a feature. In the case of namespace overlaps such as this, the ObservableProperty would override the Unobservable in the SiLA client. By setting the display_name explicitly we avoid this namespace collision.

Our implementation might be as follows:

class ClockController(sila.Feature):
    def __init__(self):
        ...
        self.time_event = asyncio.Event()
        self.subscribers = 0
        self.subscription: typing.Optional[asyncio.Task] = None
        self.current_time: typing.Optional[datetime.datetime] = None

    async def _update_current_timestamp(self):
        self.current_time = await self.get_current_timestamp()
        self.time_event.set()

    @sila.ObservableProperty(display_name="Running Clock")
    async def subscribe_current_timestamp(self) -> sila.Stream[datetime.datetime]:
        if not self.subscription:
            self.subscription = sila.utils.set_interval(self._update_current_timestamp, delay=1)

        self.subscribers += 1

        try:
            while True:
                await self.time_event.wait()
                self.time_event.clear()
                yield self.current_time
        finally:
            self.subscribers -= 1

            if self.subscribers == 0 and self.subscription:
                sila.utils.clear_interval(self.subscription)
                self.subscription = None

To ensure that only one timer runs regardless of the number of clients subscribing to the observable property, we can manage the interval at the feature level. This approach prevents multiple timers from running simultaneously and conserves resources by stopping the interval when no clients are subscribed. Therefore, we will need to modify our init to include some class-level attributes, as well as adding a method that sets the current time as a attribute after calling our unobservable property get_current_timestamp. Then we can set up a subscription with sila.utils.set_interval which will call our _update_current_timestamp function at an interval with the delay provided. This means that _update_current_timestamp will be called every second and subscribe_current_timestamp will update at this interval.

UnobservableCommand Implementation

Unobservable commands differ from properties in that they may accept parameters. In this way, they can be used to retrieve information which can be modified by the parameter given, such as in our get_current_timestamp_for_timezone function.

@sila.UnobservableCommand()
@sila.Response(name="Time Difference")
async def get_time_difference(self, future_time: datetime.time) -> int:
    """
    Get the difference in seconds between given `future_time` and the current time.

    .. parameter:: The future time to compare the current time with.
    .. return:: The time difference in in seconds.
    """

    now = datetime.datetime.now(self._tz)
    if future_time.tzinfo != self._tz:
        print(f"Replacing timezone with currently configured timezone {self._tz}.")
        future_time = future_time.replace(tzinfo=self._tz)
    future_dt = now.replace(
        hour=future_time.hour,
        minute=future_time.minute,
        second=future_time.second,
        microsecond=future_time.microsecond,
    )
    diff = future_dt - now
    return diff.seconds

Notice how this function subtly differs from get_current_timestamp in that it uses the provided timezone parameter when getting the current time, rather than using the internally configured timezone for our feature.