Controls
Controls are always SiLA Commands. They impact the state of the connector and the attached hardware or software by triggering a discrete action on the target system of the connector, e.g. to change configuration settings, or start a parameterized process.
A control allows input parameters for execution and optionally either returns just a single response (UnobservableCommand) or a stream of responses consisting of intermediate responses and a final response (ObservableCommand).
Control actions may also have additional metadata attached, such as progress information, time remaining, and execution state. Unlike properties, there are no prefixes for commands.
Intermediate Responses
During a long-running processes observable commands may wish to provide users with intermediate responses as well as status updates. The return of intermediate responses requires status: sila.Status and intermediate: sila.Intermediate[ T ] to be passed as parameters, where T is a valid SiLA data type.
Intermediate Response types cannot be inferred and must be annotated in the intermediate parameter. The IntermediateResponses of an observable command can be annotated with the sila.IntermediateResponse decorator or with .. yield:: statements in the docstring.
@sila.ObservableCommand()
@sila.IntermediateResponse(name="Current Iteration")
@sila.Response(name="Random Number")
async def iterative_method(
self, seed: int, iterations: int, *, status: sila.Status, intermediate: sila.Intermediate[int]
) -> float:
"""
Creates a random number based on a `seed` after `iterations`.
.. parameter:: Number used to initialize the pseudorandom number generator.
.. parameter:: How many times to iterate before number generation.
.. yield:: The current iteration.
.. return:: A random number.
"""
import asyncio
import random
random.seed(seed)
total = iterations + 1
for x in range(1, total):
intermediate.send(x)
status.update(progress=(x / total))
await asyncio.sleep(1)
return random.random()
Intermediate response inner types must be valid SiLA Data Types. See the Complex Data Structures Guide for more information about Custom SiLA data types.
Implementation Example
Continuing with our example of a ClockController, let's implement some Controls.
Unobservable Command
Unobservable commands can be used to modify the state of a device or controller. For our ClockController, which is initialized with no default timezone, we can create a method that allows the user to update the currently set timezone.
@sila.UnobservableCommand()
async def set_timezone(self, hours: int) -> None:
"""
Set the timezone of the current location.
.. parameter:: The offset of the timezone in hours.
"""
self._tz = datetime.timezone(datetime.timedelta(hours=hours))
Now calls to our UnobservableProperty get_current_timestamp returns a timestamp produced using the user-defined timezone.
Observable Command
An observable command for our ClockController could be a timer functionality.
We will allow the user to set a timer with a desired timeout and track as the timer winds down before finally reporting that the timer is over.
@sila.ObservableCommand()
@sila.IntermediateResponse(name="Seconds Remaining")
async def set_timer(
self,
wait_time: datetime.time,
*,
status: sila.Status,
intermediate: sila.Intermediate[int],
) -> None:
"""
Start a timer.
.. parameter:: Amount of time for timer to run.
.. yield:: Amount of seconds left before the alarm goes off.
"""
if wait_time.tzinfo != self._tz:
wait_time = wait_time.replace(tzinfo=self._tz)
td = datetime.timedelta(
hours=wait_time.hour, minutes=wait_time.minute, seconds=wait_time.second, microseconds=wait_time.microsecond
)
self._end_time: datetime.datetime = await self.get_current_timestamp() + td
while (current_time := await self.get_current_timestamp()) <= self._end_time:
# once per second we check the time and report time remaining
time_diff = self._end_time - current_time
time_remaining = time_diff.seconds
intermediate.send(time_remaining)
await asyncio.sleep(1)