Expose Python Classes

Subclass from Thing

Normally, the hardware is interfaced with a computer through Ethernet, USB, an embedded module etc., and one would write a class to encapsulate its properties and commands. Exposing this class to the network or other processes provides access to the hardware for multiple use cases in a client-server model. Such remotely visible Python objects are to be made by subclassing from Thing:

from hololinked.core import Thing, Property, action, Event


class OceanOpticsSpectrometer(Thing):
    """
    add class doc here
    """

    def __init__(self, id, serial_number, autoconnect, **kwargs):
        super().__init__(id=id, **kwargs)
        self.serial_number = serial_number
        if autoconnect:
            self.connect()

    def connect(self):
        """
        implement device driver logic/hardware communication protocol here
        to connect to the hardware
        """
        pass

id is a unique name recognising the instantiated object, allowing multiple instances of the same class to have a remote interface. It is therefore a mandatory argument to be supplied to the Thing parent. id should be a URI compatible string; non-experts may use strings composed of characters, numbers, forward slashes etc., which look like a part of a browser URL:

if __name__ == "__main__":
    spectrometer = OceanOpticsSpectrometer(
        id="spectrometer",
        serial_number="S14155",
        autoconnect=True,
        log_level=logging.DEBUG,
    )
    spectrometer.run_with_http_server(port=3569)

Properties

For attributes (like serial number above), if one requires them to be exposed on the network, one should use "properties" defined in hololinked.core.properties to "type define" the attributes of the object (in a python-idiomatic sense):

from hololinked.core import Thing, Property, action, Event
from hololinked.core.properties import Number, Selector, String, List

class OceanOpticsSpectrometer(Thing):
    """
    Spectrometer example object
    """

    serial_number = String(
        default=None, allow_None=True, doc="serial number of the spectrometer"
    )  # type: str, optional native type annotation

    def __init__(self, id, serial_number, autoconnect, **kwargs):
        super().__init__(id=id, serial_number=serial_number, **kwargs)
        # you can also pass properties to init to auto-set (optional)

Apart from predefined attributes like String, Number, List etc., it is possible to create custom properties with pydantic or JSON schema. One could also use python native types with pydantic. Only properties defined in hololinked.core.properties or subclass of Property object (note the captial 'P') can be exposed to the network, not normal python attributes or python's own property.

Actions

For methods to be exposed on the network, one can use the action decorator:

from hololinked.core import Thing, Property, action, Event

class OceanOpticsSpectrometer(Thing):
    """
    Spectrometer example object
    """

    def __init__(self, id, serial_number, autoconnect, **kwargs):
        super().__init__(id=id, serial_number=serial_number, **kwargs)
        # you can also pass properties to init to auto-set (optional)
        if autoconnect and self.serial_number is not None:
            self.connect(trigger_mode=0, integration_time=int(1e6))
            # let's say, by default

    @action()
    def connect(self, trigger_mode, integration_time):
        self.device = Spectrometer.from_serial_number(self.serial_number)
        if trigger_mode:
            self.device.trigger_mode(trigger_mode)
        if integration_time:
            self.device.integration_time_micros(integration_time)

Properties usually model settings, captured data etc., which have a read-write operation (also read-only or read-write-delete operations) and usually a specific type. Actions are supposed to model activities in the physical world, like executing a control routine, start/stop measurement etc.

Both properties and actions are symmetric - they can be invoked from within the object and externally by a client and expected to behave similarly, except when they are constrained by a state machine.

Actions can take arbitrary signature or the arguments can be constrained again using pydantic or JSON schema.

Serve the Object

To start a server, say a HTTP server, one can call the run_with_http_server method after instantiating the Thing:

if __name__ == "__main__":
    spectrometer = OceanOpticsSpectrometer(
        id="spectrometer",
        serial_number="S14155",
        autoconnect=True,
        log_level=logging.DEBUG,
    )
    spectrometer.run_with_http_server(port=3569)

The exposed properties, actions and events (events are discussed below) are independent of protocol implementations, therefore, one can start one or multiple protocols to serve the Thing:

if __name__ == "__main__":
    spectrometer = OceanOpticsSpectrometer(
        id="spectrometer", serial_number=None, autoconnect=False
    )
    spectrometer.run(
        access_points=[
            ("HTTP", 3569),
            ("ZMQ", "IPC"),
            ("MQTT", "localhost:1883"),
        ]
    )

See the protocols section for more options to serve the Thing.

All requests to properties and actions are generally queued as the domain of operation under the hood is remote procedure calls (RPC) mediated completely by ZMQ. Therefore, only one request is executed at a time as it is assumed that the hardware normally responds to only one (physical-)operation at a time.

This is only an assumption to simplify the programming model, given multiple protocols and to avoid unintended race conditions, both logical and in the physical world. One could override them explicitly using threaded or async methods.

It is also expected that the internal state of the python object is not inadvertently affected by running multiple requests at once to different properties or actions. If a single request or operation takes 5-10ms, one can still run 100s of operations per second. More often than not, the requirement of parallel operations is never the bottleneck in hardware control.

Overloaded Properties

To overload the get-set of properties to directly apply property values onto devices, one may supply a custom getter & setter method:

class Axis(Thing):
    """
    Represents a single axis of stepper module controlling a linear stage
    """

    def execute(self, command):
        # implement device driver logic to send command to hardware
        ...

    referencing_run_frequency = Number(
        bounds=(0, 40000),
        inclusive_bounds=(False, True),
        step=100,
        doc="""Run frequency during initializing (referencing), in Hz (integer value).
            I1AM0x: 40 000 maximum, I4XM01: 4 000 000 maximum""",
    )

    @referencing_run_frequency.getter
    def get_referencing_run_frequency(self):
        resp = self.execute("P08R")
        return int(resp)

    @referencing_run_frequency.setter
    def set_referencing_run_frequency(self, value):
        self.execute("P08S{}".format(value))

Properties follow the python descriptor protocol. In non expert terms, when a custom get-set method is not provided, properties look like class attributes however their data containers are instantiated at object instance level by default. For example, the serial_number property defined previously as String, whenever set/written, will be complied to a string and assigned as an attribute to each instance of the Thing class. This is done with an internally generated name. It is not necessary to know this internally generated name as the property value can be accessed again in any python logic using the dot operator, say,
self.device = Spectrometer.from_serial_number(self.serial_number)

However, to avoid generating such an internal data container and instead apply the value on the device, one must supply custom get-set methods. This is generally useful as the hardware is a better source of truth about the value of a property. Further, the write value of a property may not always correspond to a read value due to hardware limitations. Say, the write value of referencing_run_frequency requested by the user is 1050, however, the device adjusted it to 1000 automatically. This is dependent on hardware behaviour.

Publish Events

Events can asynchronously push data to clients. For example, one can supply clients with the measured data using events:

from hololinked.core import Thing, Property, action, Event

class OceanOpticsSpectrometer(Thing):
    """
    Spectrometer example object
    """
    measurement_event = Event(
        name="intensity-measurement-event",
        doc="""event generated on measurement of intensity, 
            max 30 per second even if measurement is faster.""",
    )

    def capture(self):
        self._run = True
        while self._run:
            self._intensity = self.device.intensities(
                correct_dark_counts=False, correct_nonlinearity=False
            )
            self.measurement_event.push(self._intensity.tolist())
            self.logger.debug("pushed measurement event")

Data may also be polled by the client repeatedly but events save network time or allow sending data which cannot be timed, like alarm messages. Arbitrary payloads are supported, as long as the data is serializable and one can also specify the payload structure using pydantic or JSON schema.

Events follow a PUB-SUB model, through any protocol, despite being broker-mediated like MQTT or brokerless through HTTP server sent events or ZMQ pub-sub. They follow a different channel compared to properties and actions and are not blocked by them. The events are emitted very close to execution to the push(), usually only with microseconds of delay.

To start the capture method defined above which will publish the events, one may thread it as follows to send it to the background:

class OceanOpticsSpectrometer(Thing):
    """
    Spectrometer example object
    """

    def capture(self):
        self._run = True
        while self._run:
            self._intensity = self.device.intensities(
                correct_dark_counts=False, correct_nonlinearity=False
            )
            self.measurement_event.push(self._intensity.tolist())
            self.logger.debug("pushed measurement event")

    @action()
    def start_acquisition(self):
        if self._acquisition_thread is None:
            # _acquisition_thread defined in __init__
            self._acquisition_thread = threading.Thread(target=self.capture)
            self._acquisition_thread.start()