Skip to content

Actions

API Reference

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

Actions
class GentecOpticalEnergyMeter(Thing):
    """
    Control Gentec EO optical energy meters through serial interface using this class. 
    """

    @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")

    # 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 get_number_as_instruction(cls, value):
        """
        convert a given float or int into a string of 8 characters. 
        This function is for internal use. 
        """

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

Arguments are loosely typed and may need to be constrained with a schema based on the robustness the developer is expecting in their application:

Input Schema
class GentecOpticalEnergyMeter(Thing):
    """
    Control Gentec EO optical energy meters through serial interface using this class. 
    """
    # action with input schema
    @action(
        input_schema={
            'type': 'string', 
            'enum': ['QE25LP-S-MB', 'QE12LP-S-MB-QED-D0']
        }
    )
    def set_sensor(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](instance_name='sensor')
        sensor.configure_meter(self)
        self._attached_sensor = sensor
Input Schema with Multiple Arguments
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):
    """
    6000 series picoscopes using 6000 driver from picosdk.
    """
    @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.
        """
Input Schema

Input Schema with Multiple Arguments

However, a schema is optional and it only matters that the method signature is matching when requested from a client. To enable this, set global attribute allow_relaxed_schema_actions=True. This setting is used especially when a schema is useful for validation of arguments but not available - not for methods with no arguments.

Relaxed or Unavailable Schema for Actions
1
2
3
4
5
6
class GentecOpticalEnergyMeter(Thing):
    """
    Control Gentec EO optical energy meters through serial interface using this class. 
    """

    allow_relaxed_schema_actions = True

The return value must be validated by the clients themselves. While a schema for the return value can be supplied, there is no separate validation performed on the server:

Output Schema
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,
    'items' : {
        'type': 'number',
    }
}

class Picoscope6000(Picoscope):
    """
    6000 series picoscopes using 6000 driver from picosdk.
    """
    @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]:
        v_max = ctypes.c_float()
        v_min = ctypes.c_float()
        v_range = ps.PS6000_RANGE['PS6000_{}'.format(voltage_range.upper())]
        coupling = ps.PS6000_COUPLING['PS6000_{}'.format(coupling.upper())]
        self._status['getAnalogueOffset'] = ps.ps6000GetAnalogueOffset(
                                self._ct_handle, v_range, coupling, 
                                ctypes.byref(v_max), ctypes.byref(v_min))
        assert_pico_ok(self._status['getAnalogueOffset'])
        return v_max.value, v_min.value

It is always possible to custom validate the arguments after invoking the action:

Custom Validation
from hololinked.param import ParameterizedFunction

class OceanOpticsSpectrometer(Thing):
    """
    Test object for testing the server
    """

    def __init__(self, instance_name, **kwargs):
        super().__init__(instance_name=instance_name, **kwargs)
        self.last_intensity = numpy.array([0 for i in range(1024)])

    last_intensity = ClassSelector(default=None, allow_None=True, 
                    class_=numpy.ndarray, 
                    doc="last measurement intensity (in arbitrary units)")

    @action()
    def subtract_custom_background(self, custom_background):
        if not isinstance(custom_background, numpy.ndarray):
            raise TypeError("custom_background must be a numpy array")
        return self.last_intensity - custom_background

The last and least preferred possibility is to use ParameterizedFunction:

Parameterized Function
class OceanOpticsSpectrometer(Thing):
    """
    Test object for testing the server
    """

    def __init__(self, instance_name, **kwargs):
        super().__init__(instance_name=instance_name, **kwargs)
        self.last_intensity = numpy.array([0 for i in range(1024)])

    last_intensity = ClassSelector(default=None, allow_None=True, 
                    class_=numpy.ndarray, 
                    doc="last measurement intensity (in arbitrary units)")

    @action() 
    class subtract_custom_background(ParameterizedFunction):
        """Test function with return value"""

        custom_background = ClassSelector(default=None, allow_None=True, 
                                class_=numpy.ndarray, 
                                doc="""background intensity to subtract 
                                    from the last measurement intensity
                                    (in arbitrary units)""") 

        def __call__(self, 
                    instance : "OceanOpticsSpectrometer", 
                    custom_background : numpy.ndarray
                ) -> numpy.ndarray:
            return instance.last_intensity - custom_background

ParameterizedFunction(s) are classes that implement the __call__ method and whose arguments are type defined using the same objects as properties. However, this type definition using Property object do not make these properties of the Thing. The implementation follows convention used by param where the properties are termed as "parameters" (also hence the word "ParameterizedFunction").

The __call__ method signature accepts its own self as the first argument, followed by the Thing instance as the second argument and then the arguments supplied by the client. On the client side, there is no difference between invoking a normal action and an action implemented as ParameterizedFunction:

Custom Validation
1
2
3
4
5
6
7
def server():
    O = OceanOpticsSpectrometer(
        instance_name='spectrometer',
        serial_number='S14155',
        zmq_serializer='pickle' 
    )
    O.run(zmq_protocols='IPC')
Custom Validation
1
2
3
4
5
6
7
def client():
    client = ObjectProxy(instance_name='spectrometer', protocol='IPC', 
                        serializer='pickle')
    ret = client.subtract_custom_background(
        custom_background=numpy.array([1 for i in range(len(client.last_intensity))]))
    print("reply", type(ret) , ret)
    # prints - reply <class 'numpy.ndarray'> [-1 -1 -1 ... -1 -1 -1]