Expose Python Classes

Subclass from Thing

Normally, the hardware is interfaced with a computer through Ethernet, USB etc. or any OS supported method, and one would write a class to encapsulate its properties and commands. Exposing this class to other processes and/or to the network 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.server 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. It allows multiple hardware of the same type to be connected to the same computer without overlapping the exposed interface. It is therefore a mandatory argument to be supplied to the Thing parent. Non-experts may use strings composed of characters, numbers, forward slashes etc., which looks like a part of a browser URL, but the general definition is that id should be a URI compatible string:

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.server.properties to "type define" the attributes of the object (in a python sense):

from hololinked.server import Thing, Property, action, Event
from hololinked.server.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

    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. Only properties defined in hololinked.server.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.server 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, read-write-delete operations) and 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 while using a state machine).

Actions can take arbitrary signature.

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 (discussed below) are independent of protocol implementation, 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(zmq_protocols="IPC") 
    # ZMQ interprocess-communication - suitable for beginners and 
    # apps automatically behind firewall

Further, all requests are 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 the hardware normally responds to only one operation at a time (unless one is using some hardware protocol like modbus to talk to the hardware). Further, 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. This can be overcome on need basis manually through threading or async methods. If a single request or operation takes 5-10ms, one can still run 100s of operations per second.

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 OceanOpticsSpectrometer(Thing):
    """
    Spectrometer example object 
    """

    integration_time = Number(default=1000, bounds=(0.001, 1e6), 
                doc="""integration time of measurement in milliseconds,
                        1μs (min) or 1s (max)""", crop_to_bounds=True)

    @integration_time.setter 
    def set_integration_time(self, value : float):
        self.device.integration_time_micros(int(value*1000))
        self._integration_time = int(value) 

    @integration_time.getter 
    def get_integration_time(self) -> float:
        try:
            return self._integration_time
        except:
            return 1000.0

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 OceanOpticsSpectrometer 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, 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 may supply custom get-set methods using the fget and fset argument. 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 integration_time requested by the user is 1000.2, however, the device adjusted it to 1000.0 automatically.

Push Events

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

from hololinked.server 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(f"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.

To start the capture method defined above, to receive the events, one may thread it as follows:

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

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

    @action()
    def stop_acquisition(self):
        if self._acquisition_thread is not None:
            self.logger.debug(f"""stopping acquisition thread with 
                            thread-ID {self._acquisition_thread.ident}""")
            self._run = False # break infinite loop
            self._acquisition_thread.join()
            self._acquisition_thread = None