Adding an observable command
Adding an observable command to open the door of the thermocycler.
Properties are defined to access states or measurements from a device. Commands impact the state of the connector and the attached hardware or software. A method that opens or closes the door of our simulated thermocycler is a good example of a command.
Creating the base method
As always, we start by adding the abstract method to the base class:
@abc.abstractmethod
@sila.ObservableCommand()
@sila.IntermediateResponse(name="Process status")
@sila.Response(name="Door open")
@sila.Response(name="End of process status")
async def open_door(self, *, status: sila.Status, intermediate: sila.Intermediate[str]) -> tuple[bool, str]:
"""
The abstract method to open the door.
.. yield:: The current status of the door opening process as a string.
.. return:: The state of the door where True means open and False means closed.
.. return:: The final status of the door as a string.
"""
Code explanation
For commands, we do not only write the docstring, but we also use @sila.Response decorators for documentation. The name of the responses must be defined with the respective parameters of the decorator (name=). The same applies to the @sila.IntermediateResponse, an optional feature of observable commands. The return and yield values should be described in the docstring as shown (.. return::).
For this command, we want to continuously provide the user with status updates (estimated time to completion) and an intermediate response. We will study the specifics of the implementation during the implementation class definition. The return of intermediate responses requires a status: sila.Status and an intermediate: sila.Intermediate[ str ] passed as parameters. These come after a * which requires them to be keyword-only (named) arguments (not crucial for further implementation, if needed more information can be found in the Python Documentation > Function Definitions).
Finally, we decide to return -> tuple[ bool, str ]: as we plan to return the door state after execution (bool) and an "End of process status" (human-readable string). We already defined the documentation for these return values in the @sila.Response decorators. Keep in mind that the @sila.Response decorators need to be called in the same sequence as the return types are defined in the type hinting tuple. Inside the docstring (""" ... """) we not only describe the use of the method but also use .. yield::, .. return:: and :display_name: to document the expected return values. This is technically redundant information as the @sila.Response decorator already defines these properties, however, the docstring is used by code editors to display more information. There we include this information in the docstring for a better usability.
Implementing the open_door command
Before implementing the open_door command, we import the DoorMalfunctionError, which we created in the door_controller_base.py file (see Defining a new error). We can merge the two imports as the error is defined in the same file as the base class.
from .features.door_controller import DoorControllerBase, DoorMalfunctionError
The actual step-by-step implementation:
In case the door is already open, we can finish the execution directly and return 'Door already open' as the response string:
async def open_door(self, *, status, intermediate):
if self._door_open:
status.update(
progress=1,
remaining_time=sila.datetime.timedelta(0),
)
intermediate.send('Door already open')
return self._door_open, 'Door already open'
Code explanation
With the if statement, we check if the door is open. If that is the case, using status.update, we set the progress to 1 (=100%) and the remaining time to 0. Moreover, we send intermediate return containing Door already open. While not technically necessary, these two steps provide all the status and intermediate values defined in the base class. Then we return the appropriate response string and the state of _door_open (which will be True).
As we don't interact with an actual device, we will make the "hardware error" occur randomly in the open_door method with a chance of 1 in 10 (to use this code, don't forget to import random at the top of the file).
random_int = random.randint(1, 10)
if random_int == 1:
raise DoorMalfunctionError
Code explanation
Whenever the randomly chosen integer is 1, we throw the previously defined DoorMalfunctionError. This returns the docstring from that error definition to the user and informs them about potential debugging steps.
The rest of the method handles the opening of the door. Again, as we don't interact with an actual thermocycler, we implement a five-second wait to simulate a door opening.
intermediate.send("Door opening")
for i in range(5):
status.update(
progress=i / 5,
remaining_time=sila.datetime.timedelta(seconds=5 - i),
)
await asyncio.sleep(1)
intermediate.send("Done")
status.update(
progress=1,
remaining_time=sila.datetime.timedelta(0),
)
self._door_open = True
self._door_open_change_event.set()
return self._door_open, "Door opened successfully."
Code explanation
Before we start the "door opening" process, we update the intermediate by calling intermediate.send('Door opening'). This information will be returned to the client and helps the user understand what is happening.
The door opening is implemented with a for loop executing five times. Every time, we update the status to show the progress (%) and the remaining time (s). Then we wait for 1 second using the asyncio library (remember to import asyncio).
After finishing the loop, we update the intermediate to read 'Done' and update the status to be 100 % progress with 0 seconds remaining.
As the door is now open, we change the _door_open property to true. We have set the change event as we changed the _door_open property. This makes sure that if there is an active subscription, it will note the change and update. Then we return the string and the _door_open (which is always True at this point).
Testing
To test this command's functionality, we restart the connector application and the SiLA Browser again, and we should see new the command. Before executing it, we will use the previously defined subscription method to subscribe to _door_open. If we now execute the open_door command, we should be able to observe all the intermediates, status updates, and the final return string and boolean (or the error), as well as the _door_open subscription updating.
Implementing the close_door command
If all of this works, we can finish this tutorial by defining a door_close command. You can try implementing it yourself or look at the following code block. The implementation is logically the same as the open door command. The creation of the base class is not shown here.
async def close_door(self, *, status, intermediate):
if not self._door_open:
status.update(
progress=1,
remaining_time=sila.datetime.timedelta(0),
)
intermediate.send("Door already closed")
return self._door_open, "Door already closed"
random_int = random.randint(1, 10)
if random_int == 1:
raise DoorMalfunctionError
intermediate.send("Door closing")
for i in range(5):
status.update(
progress=i / 5,
remaining_time=sila.datetime.timedelta(seconds=5 - i),
)
await asyncio.sleep(1)
intermediate.send("Done")
status.update(
progress=1,
remaining_time=sila.datetime.timedelta(0),
)
self._door_open = False
self._door_open_change_event.set()
return self._door_open, "Door closed successfully."