Skip to content

Events

API Reference

Events are pushed to the client through a publish-subscribe mechanism. Call push() method on the event to publish data to clients:

Definition
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
class GentecMaestroEnergyMeter(Thing):
    """
    Simple example to implement acquisition loops and events
    to push the captured data. Customize it for your application or 
    implement your own.
    """

    data_point_event = Event(friendly_name='data-point-event', 
                            doc='Event raised when a new data point is available',
                            label='Data Point Event')

    def loop(self):
        self._run = True
        while self._run:
            self._last_measurement = self.current_value
            timestamp = datetime.datetime.now().strftime("%H:%M:%S")
            self.data_point_event.push(dict(
                    timestamp=timestamp, 
                    energy=self._last_measurement
                ))

One can subscribe to the event using the attribute name:

Subscription
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from hololinked.client import ObjectProxy

# events works also through inter-process communication, apart from TCP
energy_meter_proxy = ObjectProxy(instance_name='gentec-maestro', protocol='IPC') 

def event_cb(event_data):
    print(event_data)

energy_meter_proxy.subscribe_event(name='data_point_event', 
                                callbacks=event_cb)

One can also supply multiple callbacks which may called in series or threaded:

Providing Callbacks
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
def event_cb1(event_data):
    print(event_data)

def event_cb2(event_data):
    print("Second callback", event_data)

energy_meter_proxy.subscribe_event(
                                name='statistics_event', 
                                callbacks=[event_cb1, event_cb2]         
                            )
Providing Callbacks
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def event_cb1(event_data):
    print(event_data)

def event_cb2(event_data):
    print("Second callback", event_data)

energy_meter_proxy.subscribe_event(
                                name='statistics_event', 
                                callbacks=[event_cb1, event_cb2], 
                                thread_callbacks=True
                            )

Schema may be supplied for the validation of the event data on the client:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class GentecMaestroEnergyMeter(Thing):

    data_point_event_schema = {
        "type": "object",
        "properties": {
            "timestamp": {"type": "string"},
            "energy": {"type": "number"}
        },
        "required": ["timestamp", "energy"]
    }

    data_point_event = Event(doc='Event raised when a new data point is available',
                            label='Data Point Event', schema=data_point_event_schema)

There is no separate validation on the server side.