Wrappers

Wrappers are the interface between the lab orchestrator and the actual devices. They translate the high-level commands defined in PythonLab to specific SiLA commands. Wrappers need to contain the code that calls the SiLA servers.

The wrapper structure

class MyWrapper(DeviceInterface):
    @staticmethod
    def get_SiLA_handler(
        step: ProcessStep, 
        cont: ContainerInfo,
        sila_client: ShakerClient,
        **kwargs,
    ) -> Observable:
        ...

All wrappers need to inherit from DeviceInterface, which for now needs to be copied from this repository.

Examples

Basic example

GreeterWrapper connects to a simple SiLA server that prints a greeting whenever it is called.

GreeterWrapper

import logging

from laborchestrator.engine.worker_interface import ObservableProtocolHandler
from laborchestrator.structures import ProcessStep, ContainerInfo

try:
    from sila2_example_server import Client as GreeterClient
except ModuleNotFoundError:
    logging.warning("The sila example server seems to not be installed")
    from sila2.client import SilaClient as GreeterClient
from . import DeviceInterface


class GreetingWrapper(DeviceInterface):
    @staticmethod
    def get_SiLA_handler(
        step: ProcessStep, cont: ContainerInfo, sila_client: GreeterClient, **kwargs
    ) -> ObservableProtocolHandler:
        # since GreetingProvider.SayHello is no observable command, we have to wrap in into an ObservableProtocolHandler
        class GreetingHandler(ObservableProtocolHandler):
            response = "None"

            def _protocol(self, client, **kwargs):
                self.response = sila_client.GreetingProvider.SayHello(cont.name)

            def get_responses(self):
                return self.response.Greeting

        handler = GreetingHandler()
        handler.run_protocol(client=None)
        return handler

Human interaction

HumanWrapper shows how human interaction can be integrated into your workflows. Interacts with the Silafied Human server

HumanWrapper

from laborchestrator.structures import ProcessStep, ContainerInfo
from . import DeviceInterface

try:
    from human_server.generated.client import Client as HumanClient
except ModuleNotFoundError:
    from sila2.client import SilaClient as HumanClient
from sila2.client import ClientObservableCommandInstance
from sila2.framework import SilaAnyType


class HumanWrapper(DeviceInterface):
    @staticmethod
    def get_SiLA_handler(
        step: ProcessStep, cont: ContainerInfo, human_client: HumanClient, **kwargs
    ) -> ClientObservableCommandInstance:
        if step.function == "ask_for_ok":
            return human_client.HumanController.CustomCommand(
                Description="Say OK",
                ResponseStructure=SilaAnyType(
                    type_xml="<DataType><Basic>Real</Basic></DataType>", value=1
                ),
            )

        elif step.function == "do_task":
            return human_client.HumanController.CustomCommand(
                Description=kwargs["message"],
                ResponseStructure=SilaAnyType(
                    type_xml="<DataType><Basic>Real</Basic></DataType>", value=1
                ),
            )

        elif step.function == "request_number":
            return human_client.HumanController.CustomCommand(
                Description=kwargs["message"],
                ResponseStructure=SilaAnyType(
                    type_xml="<DataType><Basic>Integer</Basic></DataType>", value=20
                ),
            )
        else:
            raise ValueError(f"{step.function} is unknown.")