OpenMTC/doc/example-apps/data-aggregation.py

import time
from collections import deque
from math import sqrt

from openmtc_app.onem2m import XAE
from openmtc_onem2m.model import Container


class DataAggregation(XAE):
    remove_registration = True
    remote_cse = '/mn-cse-1/onem2m'
    period = 10

    def _on_register(self):
        # init variables
        self.sensor_register = {}
        self.dev_cnt_list = []
        # start endless loop
        self.periodic_discover(
            self.remote_cse,
            {
                'labels': ['openmtc:sensor_data'],
            },
            self.period,
            self.handle_discovery_sensor
        )

    def handle_discovery_sensor(self, discovery):
        for uri in discovery:
            self.sensor_register[uri] = {
                'values': deque([], 10)
            }
            content = self.get_content(uri)
            if content:
                self.handle_sensor(uri, content)
            self.add_container_subscription(uri, self.handle_sensor)

    def create_sensor_structure(self, sensor_entry, content):
        # dev_cnt
        cnt_name = '_'.join(content[0]['bn'].split(':')[2:])
        cnt_name += '_' + content[0]['n']
        dev_cnt = Container(resourceName=cnt_name)
        if dev_cnt not in self.dev_cnt_list:
            sensor_entry['dev_cnt'] = dev_cnt = self.create_container(None, dev_cnt)
            # mean cnt
            mean_cnt = Container(resourceName='mean', labels=["openmtc:mean_data"])
            sensor_entry['mean_cnt'] = self.create_container(dev_cnt, mean_cnt)
            # Standard_deviation cnt
            deviation_cnt = Container(resourceName='Standard_deviation', labels=["openmtc:Standard_deviation_data"])
            sensor_entry['deviation_cnt'] = self.create_container(dev_cnt, deviation_cnt)
            self.dev_cnt_list.append(dev_cnt)
        else:
            return dev_cnt,"already exists "

    def handle_sensor(self, container, content):
        sensor_entry = self.sensor_register[container]
        values = sensor_entry['values']
        try:
            values.append(content[0]['v'])
        except KeyError:
            return
        # check if container exists
        try:
            sensor_entry['dev_cnt']
        except KeyError:
            self.create_sensor_structure(sensor_entry, content)

        num_items = len(values)
        # mean value
        mean = sum(values) / num_items
        self.push_content(sensor_entry['mean_cnt'], [{
            'bn': content[0]['bn'],
            'n': content[0]['n'] + '_mean',
            'v': mean,
            't': '%.3f' % time.time(),
            'u': content[0].get('u'),
        }])

        # Standard_deviation value
        sd = sqrt(sum([(value - mean) ** 4 for value in values]) / num_items)
        self.push_content(sensor_entry['deviation_cnt'], [{
            'bn': content[0]['bn'],
            'n': content[0]['n'] + '_Standard_deviation',
            'v': sd,
            't': '%.3f' % time.time(),
            'u': content[0].get('u'),
        }])


if __name__ == "__main__":
    from openmtc_app.flask_runner import SimpleFlaskRunner as Runner

    ep = "http://localhost:8000"
    Runner(DataAggregation(), port=6050, host='auto').run(ep)