Skip to content

Actions

API Reference

Only methods decorated with action() are exposed to clients.

Actions
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class GentecOpticalEnergyMeter(Thing):
    """
    Gentec EO optical energy meters with serial interface
    """
    @action()
    def set_current_value_as_zero_offset(self):
        """Set current value as offset for further measurements"""
        self.serial_comm_handle.execute_instruction("*SOU")

    @action()
    def clear_zero_offset(self):
        """Clear any offset for measurements, i.e. set offset to 0"""
        self.serial_comm_handle.execute_instruction("*COU")

    @action()
    @classmethod
    def ping(cls):
        """class method example as action - ping server"""
        return datetime.datetime.now().strftime("%H:%M:%S")

    # not an action, just a plain method
    def loop(self):
        """runs the measurement/monitoring loop"""

    # not an action, just a plain class method
    @classmethod
    def format_number_as_instruction_string(cls, value: str) -> str:
        """

Payload Validation

If arguments are loosely typed, the action will be invoked with given payload without any validation. One may validate them manually inside the method. However, one can also specify the expected argument schema using either JSON Schema or pydantic models:

Specify the expected type of the argument (with or without name)

Input Schema
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
class GentecOpticalEnergyMeter(Thing):
    """
    Gentec EO optical energy meters with serial interface
    """

    # action with input schema
    @action(
        input_schema={
            "type": "string",
            "enum": ["QE25LP-S-MB", "QE12LP-S-MB-QED-D0"],
        }
    )
    def set_sensor_model(self, value: str):
        """
        Set the attached sensor to the meter under control.
        Sensor should be defined as a class and added to the AllowedSensors dict.
        """
        sensor = allowed_sensors[value](id="sensor")
        sensor.configure_meter(self)
        self._attached_sensor = sensor
JSON schema seen in Thing Description 
GentecOpticalEnergyMeter.set_sensor_model.to_affordance().json()

{
    "description": "Set the attached sensor to the meter under control. Sensor should be defined as a class and added to the AllowedSensors dict.",
    "input": {
        "type": "string",
        "enum": ["QE25LP-S-MB", "QE12LP-S-MB-QED-D0"]
    },
    "synchronous": True
}

Specify the argument names under the properties field with type as object. Names not found in the properties field can be subsumed under python spread operator **kwargs if necessary (dont set additionalProperties to False in that case).

Input Schema with Multiple Arguments
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
set_channel_schema = {
    "type": "object",
    "properties": {
        "channel": {"type": "string", "enum": ["A", "B", "C", "D"]},
        "enabled": {"type": "boolean"},
        "voltage_range": {
            "type": "string",
            "enum": [
                "10mV",
                "20mV",
                "50mV",
                "100mV",
                "200mV",
                "500mV",
                "1V",
                "2V",
                "5V",
                "10V",
                "20V",
                "50V",
                "MAX_RANGES",
            ],
        },
        "offset": {"type": "number"},
        "coupling": {"type": "string", "enum": ["AC", "DC"]},
        "bw_limiter": {"type": "string", "enum": ["full", "20MHz"]},
    },
}

class Picoscope6000(Picoscope):

    @action(input_schema=set_channel_schema)
    def set_channel(
        self,
        channel: str,
        enabled: bool = True,
        v_range: str = "2V",
        offset: float = 0,
        coupling: str = "DC_1M",
        bw_limiter: str = "full",
    ) -> None:
        """
        Set the parameter for a channel.
        https://www.picotech.com/download/manuals/picoscope-6000-series-a-api-programmers-guide.pdf
        """
JSON schema seen in Thing Description 
Picoscope6000.set_channel.to_affordance().json()

{
    "description": "Set the parameter for a channel. https://www.picotech.com/download/manuals/picoscope-6000-series-a-api-programmers-guide.pdf",
    "input": {
        "type": "object",
        "properties": {
            "channel": {"type": "string", "enum": ["A", "B", "C", "D"]},
            "enabled": {"type": "boolean"},
            "voltage_range": {
                "type": "string",
                "enum": [
                    "10mV",
                    "20mV",
                    "50mV",
                    "100mV",
                    "200mV",
                    "500mV",
                    "1V",
                    "2V",
                    "5V",
                    "10V",
                    "20V",
                    "50V",
                    "MAX_RANGES"
                ]},
            "offset": {"type": "number"},
            "coupling": {"type": "string", "enum": ["AC", "DC"]},
            "bw_limiter": {"type": "string", "enum": ["full", "20MHz"]}
        }
    },
    "synchronous": True
}

Specify return type under output_schema field:

With Return Type
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
analog_offset_input_schema = {
    "type": "object",
    "properties": {
        "voltage_range": {
            "type": "string",
            "enum": [
                "10mV",
                "20mV",
                "50mV",
                "100mV",
                "200mV",
                "500mV",
                "1V",
                "2V",
                "5V",
                "10V",
                "20V",
                "50V",
                "MAX_RANGES",
            ],
        },
        "coupling": {"type": "string", "enum": ["AC", "DC"]},
    },
}

analog_offset_output_schema = {
    "type": "array",
    "minItems": 2,
    "maxItems": 2,
    "items": {
        "type": "number",
    },
}

class Picoscope6000(Picoscope):

    @action(
        input_schema=analog_offset_input_schema,
        output_schema=analog_offset_output_schema,
    )
    def get_analogue_offset(
        self, voltage_range: str, coupling: str
    ) -> typing.Tuple[float, float]:
JSON schema seen in Thing Description 
Picoscope6000.get_analogue_offset.to_affordance().json()

{
    "description": "analogue offset for a voltage range and coupling",
    "synchronous": True,
    "input": {
        "type": "object",
        "properties": {
            "voltage_range": {
                "type": "string",
                "enum": ["10mV",
                    "20mV",
                    "50mV",
                    "100mV",
                    "200mV",
                    "500mV",
                    "1V",
                    "2V",
                    "5V",
                    "10V",
                    "20V",
                    "50V",
                    "MAX_RANGES"
                ]
            },
            "coupling": {"type": "string", "enum": ["AC", "DC"]}
        }
    },
    "output": {
        "type": "array",
        "minItems": 2,
        "maxItems": 2,
        "items": {"type": "number"}
    },
}

Type annotate the argument, either plainly or with Annotated. A pydantic model will be composed with the argument name as the field name and the type annotation as the field type:

Input Schema with Single Argument
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from typing import Annotated

class GentecOpticalEnergyMeter(Thing):

    @action()
    def start_acquisition(self,
        max_count: Annotated[int, Field(gt=0)]
    ) -> None:
        """
        Start acquisition of energy measurements.

        Parameters
        ----------
        max_count: int
            maximum number of measurements to acquire before stopping automatically.
        """
JSON schema seen in Thing Description 
GentecOpticalEnergyMeter.start_acquisition.to_affordance().json()

{
    "description": "Start acquisition of energy measurements. max_count: maximum number of measurements to acquire before stopping automatically.",
    "input": {
        "properties": {
            "max_count": {
                "exclusiveMinimum": 0,
                "type": "integer"
            }
        },
        "required": ["max_count"],
        "title": "start_acquisition_input",
        "type": "object"
    },
    "synchronous": True
}

Again, type annotate the arguments, either plainly or with Annotated:

Input Schema with Multiple Arguments
from typing import Literal

class Picoscope6000(Thing):

    @action()
    def set_channel(
        self,
        channel: Literal["A", "B", "C", "D"],
        enabled: bool = True,
        v_range: Literal[
            "10mV",
            "20mV",
            "50mV",
            "100mV",
            "200mV",
            "500mV",
            "1V",
            "2V",
            "5V",
            "10V",
            "20V",
            "50V",
            "MAX_RANGES",
        ] = "2V",
        offset: float = 0,
        coupling: Literal["AC", "DC"] = "DC_1M",
        bw_limiter: Literal["full", "20MHz"] = "full",
    ) -> None:
JSON schema seen in Thing Description 
Picoscope6000.set_channel.to_affordance().json()

{
    "description": "Set the parameter for a channel. https://www.picotech.com/download/manuals/picoscope-6000-series-a-api-programmers-guide.pdf",
    "input": {
        "properties": {
            "channel": {
                "enum": ["A", "B", "C", "D"],
                "type": "string"
            },
            "enabled": {"default": True, "type": "boolean"},
            "v_range": {
                "default": "2V",
                "enum": [
                    "10mV",
                    "20mV",
                    "50mV",
                    "100mV",
                    "200mV",
                    "500mV",
                    "1V",
                    "2V",
                    "5V",
                    "10V",
                    "20V",
                    "50V",
                    "MAX_RANGES"
                ],

                "type": "string"
            },
            "offset": {"default": 0, "type": "number"},
            "coupling": {
                "default": "DC_1M",
                "enum": ["AC", "DC"],
                "type": "string"
            },
            "bw_limiter": {
                "default": "full",
                "enum": ["full", "20MHz"],
                "type": "string"
            }
        },
        "required": ["channel"],
        "type": "object"
    },
    "synchronous": True
}

type annotate the return type:

from typing import Annotated
from pydantic import Field

class SerialUtility(Thing):

    @action()
    def execute_instruction(
        self, command: str, return_data_size: Annotated[int, Field(ge=0)] = 0
    ) -> str:
        """
        executes instruction given by the ASCII string parameter 'command'. If return data size is greater than 0, it reads the response and returns the response. Return Data Size - in bytes - 1 ASCII character = 1 Byte.
        """
JSON schema seen in Thing Description 
SerialUtility.execute_instruction.to_affordance().json()

{
    "description": "executes instruction given by the ASCII string parameter 'command'. If return data size is greater than 0, it reads the response and returns the response. Return Data Size - in bytes - 1 ASCII character = 1 Byte.",
    "input": {
        "properties": {
            "command": {"type": "string"},
            "return_data_size": {"default": 0, "minimum": 0, "type": "integer"}
        },
        "required": ["command"],
        "type": "object"
    },
    "output": {"type": "string"},
    "synchronous": True
}

If the composed models from the type annotations are not sufficient or contain errors, one may directly supply the models:

With Direct Models
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from pydantic import BaseModel, field_validator

class CommandModel(BaseModel):
    command: str
    return_data_size: int = Field(0, ge=0)

    @field_validator("command")
    def validate_command(cls, v):
        if not isinstance(v, str) or not v:
            raise ValueError("Command must be a non-empty string")
        if command not in SerialUtility.supported_commands:
            raise ValueError(f"Command {command} is not supported")
        return v

class ResponseModel(BaseModel):
    response: str = Field(..., description="Response from the device")

class SerialUtility(Thing):

    supported_commands = ["*IDN?", "MEAS:VOLT?", "MEAS:CURR?"]

    @action(input_schema=CommandModel, output_schema=ResponseModel)
    def execute_instruction(self, command: str, return_data_size: int = 0) -> str:
        """
        executes instruction given by the ASCII string parameter 'command'
        """

Threaded & Async Actions

Actions can be made asynchronous or threaded by setting the synchronous flag to False. For methods that are not async, they will run in a separate thread:

Threaded Actions
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class ServoMotor(Thing):

    @action(synchronous=False)
    def poll_device_state(self) -> str:
        """check device state, especially when it got stuck up"""
        ...

    @action(threaded=True) # exactly the same effect for sync methods
    def poll_device_state(self) -> str:
        """check device state, especially when it got stuck up"""
        ...

The return value is fetched and returned to the client. One could also start long running actions without fetching a return value, (although it would be better in many cases to manually thread out a long running action):

Threaded Actions
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
class DCPowerSupply(Thing):
    """A DC Power Supply from 0-30V"""

    # The suitability of this example in a realistic use case is untested
    @action(threaded=True)
    def monitor_over_voltage(self, period: float = 5):
        """background voltage monitor loop"""
        while True:
            voltage = self.measure_voltage()
            if voltage > self.over_voltage_threshold:
                timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                self.over_voltage_event(
                    dict(
                        timestamp=timestamp,
                        voltage=voltage
                    )
                )
            time.sleep(period)

For async actions, they are scheduled in the running event loop as a task:

Async Actions
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class DCPowerSupply(Thing):

    @action(synchronous=False)
    # @action(create_task=True) # exactly the same effect for async methods
    async def monitor_over_voltage(self, period: float = 5):
        """background monitor loop"""
        while True:
            voltage = await asyncio.get_running_loop().run_in_executor(
                            None, self.measure_voltage
                    )
            if voltage > self.over_voltage_threshold:
                timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                self.over_voltage_event(
                    dict(
                        timestamp=timestamp,
                        voltage=voltage
                    )
                )
            await asyncio.sleep(period)
    # The suitability of this example in a realistic use case is untested

For long running actions that do not return, call them with oneway flag on the client, otherwise expect a TimeoutError:

client.invoke_action("monitor_over_voltage", period=10, oneway=True)

Thing Description Metadata

field supported description
input ✓ schema of the input payload (validation carried out)
output ✓ schema of the output payload (validation not carried out)
safe ✓ whether the action is safe to execute, only treated as a metadata
idempotent ✓ whether the action is idempotent, True when the action is executable in all states of a state machine, otherwise False
synchronous ✓ whether the action is synchronous, False for threaded actions and async actions which are scheduled in the running event loop